diff --git a/docs/events/background_events.md b/docs/events/background_events.md
index 73d7b1e27bc..a779a1c9997 100644
--- a/docs/events/background_events.md
+++ b/docs/events/background_events.md
@@ -196,4 +196,5 @@ Background tasks mostly work like normal `EventHandler` methods, with certain ex
 - Background tasks cannot modify the state outside of an `async with self` context block.
 - Background tasks may read the state outside of an `async with self` context block, but the value may be stale.
 - Calling another event handler as a method of `self`, including one inherited from a parent state, runs it through the same proxy, so do it inside `async with self`. In the called handler `type(self)` is `StateProxy`; use `self.__class__` for the state class. See [Upgrading to Reflex 0.10](/docs/changelog/upgrading/upgrading-to-0-10/#calling-inherited-handlers-from-background-tasks).
+- Awaiting an event handler through its state class, like `await State.increment()`, works in a background task. Outside `async with self` the event is queued and awaited, so it takes the lock on its own. Inside `async with self` a regular handler runs right away on the locked state, but a background handler raises `EventHandlerTypeError`; `yield` it instead. See [Awaiting Event Handlers](/docs/events/chaining-events/#awaiting-event-handlers).
 - Background tasks may not be directly called from other event handlers or background tasks. Instead use `yield` or `return` to trigger the background task.
diff --git a/docs/events/chaining_events.md b/docs/events/chaining_events.md
index bdbf4e1e2f9..5c0649e3428 100644
--- a/docs/events/chaining_events.md
+++ b/docs/events/chaining_events.md
@@ -47,11 +47,63 @@ def call_handler_example():
     )
 ```
 
+## Awaiting Event Handlers
+
+An event handler accessed through its state class is awaitable. `await` runs it
+right away, inside the current event, and gives back its return value. The
+awaited handler runs on the same locked state as the caller, its `yield`s send
+updates to the frontend as usual, and any backend events it yields or returns
+are queued after the current event, as if the caller had yielded them;
+frontend-only events are sent to the frontend immediately.
+
+```python demo exec id=await-handler
+class AwaitHandlerState(rx.State):
+    total: int = 0
+    last_added: int = 0
+
+    @rx.event
+    def add(self, amount: int) -> int:
+        """Add to the total and return the new total."""
+        self.total += amount
+        self.last_added = amount
+        return self.total
+
+    @rx.event
+    async def add_twice(self, amount: int):
+        first = await type(self).add(amount)
+        second = await type(self).add(amount)
+        yield rx.toast(f"{first} then {second}")
+
+
+def await_handler_example():
+    return rx.vstack(
+        rx.text(f"Total: {AwaitHandlerState.total}"),
+        rx.button("Add 5 twice", on_click=AwaitHandlerState.add_twice(5)),
+    )
+```
+
+Compared to calling the handler as a method of `self`, awaiting it converts
+the arguments like an event from the frontend, sends an update when it yields,
+and works for a handler of any state in the app, not only the caller's own
+state.
+
+```md alert info
+# Rules for awaiting event handlers
+
+- Only inside an event handler (or `app.modify_state`); awaiting a handler anywhere else raises an error.
+- Calling the handler, like `State.add(5)`, only binds its arguments. Nothing runs until it is awaited, so it can still be returned or yielded as before. Each `await` runs it again.
+- A background task awaiting a handler outside `async with self` queues the event and waits for it to finish; inside `async with self` the handler runs right away on the locked state.
+- A background handler cannot be awaited while the state lock is held, as it would wait for that lock. `yield` it instead.
+- Frontend-only events like `rx.redirect` cannot be awaited. `yield` or return them.
+- Awaiting several handlers at once with `asyncio.gather` works, but they run concurrently on the same state and interleave at every `await` inside them. Await them one at a time when order matters.
+- A task started from a handler, like `asyncio.ensure_future(State.add(1))`, keeps the state lock held until it finishes when it starts before the handler returns. If the handler returns first, the task's handler is queued like a yielded event and runs under its own lock.
+```
+
 ## Returning Events From Event Handlers
 
 So far, we have only seen events that are triggered by components. However, an event handler can also return events.
 
-In Reflex, event handlers run synchronously, so only one event handler can run at a time, and the events in the queue will be blocked until the current event handler finishes.The difference between returning an event and calling an event handler is that returning an event will send the event to the frontend and unblock the queue.
+In Reflex, event handlers run synchronously, so only one event handler can run at a time, and the events in the queue will be blocked until the current event handler finishes. The difference between returning an event and calling an event handler is that a returned event is queued and runs after the current handler finishes.
 
 ```md alert info
 # Be sure to use the class name `State` (or any substate) rather than `self` when returning events.
diff --git a/docs/events/decentralized_event_handlers.md b/docs/events/decentralized_event_handlers.md
index 5898b164a4d..552b689e6d8 100644
--- a/docs/events/decentralized_event_handlers.md
+++ b/docs/events/decentralized_event_handlers.md
@@ -159,3 +159,40 @@ def process_form(state: FormState, data: dict):
     # Process form data
     return validate_data  # Chain to another event
 ```
+
+## Stateless Event Handlers
+
+A function decorated with `@rx.event` whose first parameter is not a state is a
+stateless event handler. It is registered like any other handler, so it can be
+used as an event trigger and awaited from other handlers, but it runs without a
+state and takes no state lock of its own. Triggered from the frontend, it runs
+at once and the client's other events do not wait behind it; awaited from a
+handler, it runs inline under whatever lock that handler holds.
+
+```python box
+class PortfolioState(rx.State):
+    price: float = 0.0
+
+    @rx.event
+    def show_price(self, price: float):
+        self.price = price
+
+
+@rx.event
+async def refresh_quote(symbol: str):
+    # No state lock is held while the request is in flight.
+    yield PortfolioState.show_price(await price_service.quote(symbol))
+
+
+def quote_button():
+    return rx.button("Refresh", on_click=refresh_quote("ACME"))
+```
+
+Events a stateless handler yields or returns run against the state of the
+client that triggered it, so `refresh_quote` fetches the quote with no lock
+held and then hands the result to `PortfolioState` like any chained event.
+Without the decorator, `refresh_quote` would be a plain function that cannot be
+used as a trigger. Awaiting a stateless handler from another handler runs it
+inline, so call external services from a
+[background task](/docs/events/background-events/) or from a stateless trigger
+like this one, never from inside `async with self`.
diff --git a/news/+awaitable-event-handlers.breaking.md b/news/+awaitable-event-handlers.breaking.md
new file mode 100644
index 00000000000..8cbf88e9955
--- /dev/null
+++ b/news/+awaitable-event-handlers.breaking.md
@@ -0,0 +1 @@
+`@rx.event` now returns an `EventHandler` instead of the decorated function. A decorator stacked above `@rx.event` is deprecated: the method still runs and registers under its own name, with a deprecation warning, until 1.0; apply other decorators below `@rx.event`. `@rx.event` on a method of a class that is not a state keeps working when called through an instance, and on a `staticmethod` or `classmethod` leaves it unchanged; both warn and should be removed, and calling such a method through the class (`Helper.compute(obj, 3)`) is not supported. A module-level function decorated with `@rx.event` whose first parameter is not a state is now a stateless event handler, so calling it builds an event instead of running it; drop the decorator from plain helper functions.
diff --git a/news/+awaitable-event-handlers.feature.md b/news/+awaitable-event-handlers.feature.md
new file mode 100644
index 00000000000..a15586d2bae
--- /dev/null
+++ b/news/+awaitable-event-handlers.feature.md
@@ -0,0 +1,14 @@
+Event handlers are awaitable: `await State.handler(args)` inside another event handler runs it right away on the same locked state and returns its result, so a handler may return a value to the backend code awaiting it (returned events still chain, a returned coroutine still raises, and return values are not sent to the frontend yet). `@rx.event` returns the `EventHandler` itself, a function without a state parameter registers as a stateless handler that runs without the state lock, and `self.handler()` keeps calling the plain method as before.
+
+```python
+@rx.event
+def add(self, amount: int) -> int:
+    self.total += amount
+    return self.total
+
+
+@rx.event
+async def add_twice(self, amount: int):
+    first = await type(self).add(amount)
+    yield rx.toast(f"total is now {await type(self).add(amount)}")
+```
diff --git a/packages/reflex-base/news/+awaitable-event-handlers.breaking.md b/packages/reflex-base/news/+awaitable-event-handlers.breaking.md
new file mode 100644
index 00000000000..8cbf88e9955
--- /dev/null
+++ b/packages/reflex-base/news/+awaitable-event-handlers.breaking.md
@@ -0,0 +1 @@
+`@rx.event` now returns an `EventHandler` instead of the decorated function. A decorator stacked above `@rx.event` is deprecated: the method still runs and registers under its own name, with a deprecation warning, until 1.0; apply other decorators below `@rx.event`. `@rx.event` on a method of a class that is not a state keeps working when called through an instance, and on a `staticmethod` or `classmethod` leaves it unchanged; both warn and should be removed, and calling such a method through the class (`Helper.compute(obj, 3)`) is not supported. A module-level function decorated with `@rx.event` whose first parameter is not a state is now a stateless event handler, so calling it builds an event instead of running it; drop the decorator from plain helper functions.
diff --git a/packages/reflex-base/news/+awaitable-event-handlers.feature.md b/packages/reflex-base/news/+awaitable-event-handlers.feature.md
new file mode 100644
index 00000000000..a15586d2bae
--- /dev/null
+++ b/packages/reflex-base/news/+awaitable-event-handlers.feature.md
@@ -0,0 +1,14 @@
+Event handlers are awaitable: `await State.handler(args)` inside another event handler runs it right away on the same locked state and returns its result, so a handler may return a value to the backend code awaiting it (returned events still chain, a returned coroutine still raises, and return values are not sent to the frontend yet). `@rx.event` returns the `EventHandler` itself, a function without a state parameter registers as a stateless handler that runs without the state lock, and `self.handler()` keeps calling the plain method as before.
+
+```python
+@rx.event
+def add(self, amount: int) -> int:
+    self.total += amount
+    return self.total
+
+
+@rx.event
+async def add_twice(self, amount: int):
+    first = await type(self).add(amount)
+    yield rx.toast(f"total is now {await type(self).add(amount)}")
+```
diff --git a/packages/reflex-base/news/+literal-var-proxy-cache.bugfix.md b/packages/reflex-base/news/+literal-var-proxy-cache.bugfix.md
new file mode 100644
index 00000000000..a61a9e00867
--- /dev/null
+++ b/packages/reflex-base/news/+literal-var-proxy-cache.bugfix.md
@@ -0,0 +1 @@
+Fix a `TypeError` when an event handler is called with a proxied state value (such as `self.other(self.items)`) after one was called with a proxied value of a different kind, such as a dict.
diff --git a/packages/reflex-base/src/reflex_base/event/__init__.py b/packages/reflex-base/src/reflex_base/event/__init__.py
index 7f6c6ffe43e..344bfa5d5a6 100644
--- a/packages/reflex-base/src/reflex_base/event/__init__.py
+++ b/packages/reflex-base/src/reflex_base/event/__init__.py
@@ -8,7 +8,7 @@
 import types
 import warnings
 from base64 import b64encode
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Generator, Mapping, Sequence
 from functools import cached_property, lru_cache, partial
 from typing import (
     TYPE_CHECKING,
@@ -39,7 +39,7 @@
 from reflex_base.components.field import BaseField
 from reflex_base.constants.compiler import CompileVars, Imports
 from reflex_base.registry import RegistrationContext
-from reflex_base.utils import format
+from reflex_base.utils import console, format
 from reflex_base.utils.decorator import once
 from reflex_base.utils.exceptions import (
     EventFnArgMismatchError,
@@ -147,22 +147,7 @@ def from_event_type(
             if not isinstance(e, EventSpec):
                 msg = f"Unexpected event type, {type(e)}."
                 raise ValueError(msg)
-            name = format.format_event_handler(e.handler)
-            # Detach mutable values from any state-bound proxies (e.g.
-            # ImmutableMutableProxy from a background task's StateProxy),
-            # copying only subtrees that are actually proxied.
-            payload = {
-                k._js_expr: _detach_state_proxies(v._decode()) for k, v in e.args
-            }
-
-            # Create an event and append it to the list.
-            out.append(
-                Event(
-                    name=name,
-                    payload=payload,
-                    router_data=router_data or {},
-                )
-            )
+            out.append(e._to_event(router_data))
 
         return out
 
@@ -511,6 +496,17 @@ async def _no_chain_background_task_gen(*args, **kwargs):  # noqa: RUF029
     raise TypeError(msg)
 
 
+def _base_state_type() -> type:
+    """The BaseState class, imported lazily to avoid a circular import.
+
+    Returns:
+        ``reflex.state.BaseState``.
+    """
+    from reflex.state import BaseState
+
+    return BaseState
+
+
 @dataclasses.dataclass(
     init=True,
     frozen=True,
@@ -655,6 +651,54 @@ def supersedes(self) -> bool:
         """
         return getattr(self.fn, SUPERSEDES_MARKER, False)
 
+    @property
+    def __wrapped__(self) -> Callable:
+        """The decorated function, as decorators and proxies expose it.
+
+        Returns:
+            The function the handler runs.
+        """
+        return self.fn
+
+    def __await__(self) -> Generator[Any, None, Any]:
+        """Run the handler with no arguments in the active EventContext.
+
+        Returns:
+            The awaitable's iterator; awaiting it yields the handler's return value.
+        """
+        return self().__await__()
+
+    def __set_name__(self, owner: type, name: str) -> None:
+        """Bind the handler to the state class whose body declares it.
+
+        Runs when the class is created, before ``BaseState.__init_subclass__``
+        collects and registers the class's handlers. A handler already bound
+        to a state (aliased onto another class) is left as it is. A mixin's
+        handler is bound to the mixin, which marks it as declared there: each
+        state using the mixin gets a copy of its own.
+
+        Args:
+            owner: The class the handler is declared on.
+            name: The attribute name.
+        """
+        if self.state is not None:
+            return
+        from reflex.state import BaseState
+
+        if isinstance(owner, type) and issubclass(owner, BaseState):
+            object.__setattr__(self, "state", owner)
+            self._get_type_hints()
+            return
+        console.deprecate(
+            feature_name="@rx.event on a method of a class that is not a state",
+            reason=(
+                f"`{owner.__qualname__}.{name}` stays a plain method; remove the "
+                "decorator."
+            ),
+            deprecation_version="0.10.2",
+            removal_version="1.0",
+        )
+
     def __get__(self, instance: Any, owner: type | None = None) -> Any:
         """Get the handler on class access, or its function bound to a state.
 
@@ -667,12 +711,14 @@ def __get__(self, instance: Any, owner: type | None = None) -> Any:
             instance of the handler's state (an ancestor of ``instance`` for
             an inherited handler).
         """
-        if (
-            instance is None
-            or self.state is None
+        if instance is None:
+            return self
+        if self.state is None:
+            # A method of a class that is not a state: a plain bound method,
+            # as when @rx.event returned the function.
+            return types.MethodType(self.fn, instance)
+        if not isinstance(instance, self.state):
             # Held by a class that is not its state, nor a substate of it.
-            or not isinstance(instance, self.state)
-        ):
             return self
         state = (
             instance
@@ -700,6 +746,21 @@ def __call__(self, *args: Any, **kwargs: Any) -> "EventSpec":
         """
         from reflex_base.utils.exceptions import EventHandlerTypeError
 
+        if args and isinstance(args[0], self.state or _base_state_type()):
+            # A decorator stacked above @rx.event calls the handler like the
+            # plain method it used to be (the handler itself is never bound:
+            # the class attribute is the wrapper). Run it, as before.
+            console.deprecate(
+                feature_name="Calling an event handler with a state instance",
+                reason=(
+                    f"`{self.fn.__qualname__}` is called through a decorator "
+                    "stacked above @rx.event; apply other decorators below it."
+                ),
+                deprecation_version="0.10.2",
+                removal_version="1.0",
+            )
+            return self.fn(*args, **kwargs)
+
         # Get the function args.
         if self.state is not None:
             # Skip the `self` arg for state-bound event handlers.
@@ -831,6 +892,63 @@ def __init__(
         object.__setattr__(self, "client_handler_name", client_handler_name)
         object.__setattr__(self, "args", args or ())
 
+    def __hash__(self) -> int:
+        """Hash the spec by identity, so ``asyncio.gather`` can track it as an awaitable.
+
+        The bound args are Vars, whose ``==`` builds a Var instead of a bool,
+        so the field-wise hash and equality a dataclass generates are unusable
+        for a spec with args.
+
+        Returns:
+            The hash of the spec.
+        """
+        return id(self)
+
+    @property
+    def __wrapped__(self) -> Callable:
+        """The decorated function of the handler the spec binds arguments for.
+
+        Returns:
+            The function the handler runs.
+        """
+        return self.handler.fn
+
+    def __await__(self) -> Generator[Any, None, Any]:
+        """Run the event in the active EventContext.
+
+        Inline on the locked state tree while the current task holds the state
+        lock, else enqueued and awaited. Each await runs the event again: a
+        spec describes an event, like a ``functools.partial`` describes a call.
+
+        Returns:
+            The awaitable's iterator; awaiting it yields the handler's return value.
+        """
+        from reflex_base.event.processor.base_state_processor import (
+            execute_event_inline,
+        )
+
+        return execute_event_inline(self).__await__()
+
+    def _to_event(self, router_data: dict[str, Any] | None = None) -> Event:
+        """Create the runtime Event this spec describes.
+
+        Args:
+            router_data: The routing data for the event.
+
+        Returns:
+            The Event, with its payload decoded from the bound args.
+        """
+        return Event(
+            name=format.format_event_handler(self.handler),
+            # Detach mutable values from any state-bound proxies (e.g.
+            # ImmutableMutableProxy from a background task's StateProxy),
+            # copying only subtrees that are actually proxied.
+            payload={
+                k._js_expr: _detach_state_proxies(v._decode()) for k, v in self.args
+            },
+            router_data=router_data or {},
+        )
+
     def with_args(self, args: tuple[tuple[Var, Var], ...]) -> "EventSpec":
         """Copy the event spec, with updated args.
 
@@ -2882,12 +3000,29 @@ class EventCallback(Generic[*P], EventActionsMixin):
     """A descriptor that wraps a function to be used as an event."""
 
     if TYPE_CHECKING:
-        # EventCallback is never instantiated: `event()` returns the undecorated
-        # function, which the state metaclass turns into an EventHandler. This
-        # class is only the static stand-in for it, so declare the EventHandler
-        # attribute that `SomeState.handler` actually exposes at runtime.
+        # EventCallback is never instantiated: `event()` returns an
+        # EventHandler. This class is only the static stand-in for it, so
+        # declare the EventHandler members that `SomeState.handler` actually
+        # exposes at runtime.
         fn: Callable[[Any, Unpack[P]], Any]
 
+        @property
+        def __wrapped__(self) -> Callable[[Any, Unpack[P]], Any]:
+            """The decorated function.
+
+            Returns:
+                The function the handler runs.
+            """
+            ...
+
+        def __await__(self) -> Generator[Any, None, Any]:
+            """Run the handler in the active EventContext.
+
+            Returns:
+                The awaitable's iterator; awaiting it yields the handler's return value.
+            """
+            ...
+
     def __init__(self, func: Callable[[Any, Unpack[P]], Any]):
         """Initialize the descriptor with the function to be wrapped.
 
@@ -3119,6 +3254,26 @@ def __call__(self, *args: Var) -> Any:
     from reflex.state import BaseState
 
 
+class EventDecorator(Protocol):
+    """The decorator ``rx.event(...)`` returns, for state-bound and stateless handlers."""
+
+    @overload
+    def __call__(
+        self, func: "Callable[[BASE_STATE, Unpack[P]], Any]"
+    ) -> EventCallback[*P]: ...
+
+    @overload
+    def __call__(self, func: "Callable[[Unpack[P]], Any]") -> EventCallback[*P]: ...
+
+    def __call__(self, func: Callable) -> EventCallback:
+        """Wrap a function to be used as an event.
+
+        Args:
+            func: The function to wrap.
+        """
+        ...
+
+
 class EventNamespace:
     """A namespace for event related classes."""
 
@@ -3194,7 +3349,7 @@ def __new__(
         throttle: int | None = None,
         debounce: int | None = None,
         temporal: bool | None = None,
-    ) -> "Callable[[Callable[[BASE_STATE, Unpack[P]], Any]], EventCallback[*P]]": ...
+    ) -> EventDecorator: ...
 
     @overload
     def __new__(
@@ -3210,9 +3365,24 @@ def __new__(
         temporal: bool | None = None,
     ) -> EventCallback[*P]: ...
 
+    # A stateless handler: a function whose first parameter is not a state.
+    @overload
+    def __new__(
+        cls,
+        func: "Callable[[Unpack[P]], Any]",
+        *,
+        background: bool | None = None,
+        supersedes: bool | None = None,
+        stop_propagation: bool | None = None,
+        prevent_default: bool | None = None,
+        throttle: int | None = None,
+        debounce: int | None = None,
+        temporal: bool | None = None,
+    ) -> EventCallback[*P]: ...
+
     def __new__(
         cls,
-        func: "Callable[[BASE_STATE, Unpack[P]], Any] | None" = None,
+        func: "Callable[..., Any] | None" = None,
         *,
         background: bool | None = None,
         supersedes: bool | None = None,
@@ -3221,7 +3391,7 @@ def __new__(
         throttle: int | None = None,
         debounce: int | None = None,
         temporal: bool | None = None,
-    ) -> "EventCallback[*P] | Callable[[Callable[[BASE_STATE, Unpack[P]], Any]], EventCallback[*P]]":
+    ) -> "EventCallback | EventDecorator":
         """Wrap a function to be used as an event.
 
         Args:
@@ -3272,9 +3442,16 @@ def _build_event_actions():
                 event_actions["temporal"] = temporal
             return event_actions
 
-        def wrapper(
-            func: "Callable[[BASE_STATE, Unpack[P]], T]",
-        ) -> EventCallback[*P]:
+        def wrapper(func: Callable[..., Any]) -> EventCallback:
+            if isinstance(func, (staticmethod, classmethod)):
+                # Never an event handler; left as it is, like before.
+                console.deprecate(
+                    feature_name="@rx.event on a staticmethod or classmethod",
+                    reason="The decorator has no effect there; remove it.",
+                    deprecation_version="0.10.2",
+                    removal_version="1.0",
+                )
+                return func  # pyright: ignore [reportReturnType]
             if background is True:
                 if not inspect.iscoroutinefunction(
                     func
@@ -3288,48 +3465,49 @@ def wrapper(
                 msg = "Event handlers cannot be private."
                 raise ValueError(msg)
 
+            # The markers stay on the function: a mixin's handler is copied
+            # per concrete state from its function, and the handler reads
+            # them from there.
+            event_actions = _build_event_actions()
+            if event_actions:
+                setattr(func, EVENT_ACTIONS_MARKER, event_actions)
+
             qualname: str | None = getattr(func, "__qualname__", None)
 
             if qualname and (
                 len(func_path := qualname.split(".")) == 1
                 or func_path[-2] == "<locals>"
             ):
+                # A plain function: a decentralized handler when its first
+                # parameter is a state, else a stateless one. Either is
+                # registered now, under a module-qualified name.
                 from reflex.state import BaseState
 
                 types = get_type_hints(func)
                 state_arg_name = next(iter(inspect.signature(func).parameters), None)
                 state_cls = state_arg_name and types.get(state_arg_name)
-                if state_cls and issubclass(state_cls, BaseState):
-                    name = (
-                        (func.__module__ + "." + qualname)
-                        .replace(".", "_")
-                        .replace("<locals>", "_")
-                        .removeprefix("_")
-                    )
-                    object.__setattr__(func, "__name__", name)
-                    object.__setattr__(func, "__qualname__", name)
+                name = (
+                    (func.__module__ + "." + qualname)
+                    .replace(".", "_")
+                    .replace("<locals>", "_")
+                    .removeprefix("_")
+                )
+                object.__setattr__(func, "__name__", name)
+                object.__setattr__(func, "__qualname__", name)
+                if isinstance(state_cls, type) and issubclass(state_cls, BaseState):
                     state_cls._add_event_handler(name, func)
-                    event_callback = getattr(state_cls, name)
-
-                    # Apply decorator event actions
-                    event_actions = _build_event_actions()
-                    if event_actions:
-                        # Create new EventCallback with updated event_actions
-                        event_callback = dataclasses.replace(
-                            event_callback, event_actions=event_actions
-                        )
-
-                    return event_callback
+                    return getattr(state_cls, name)
+                return RegistrationContext.register_event_handler(
+                    EventHandler(fn=func, event_actions=event_actions)
+                )  # pyright: ignore [reportReturnType]
 
-            # Store decorator event actions on the function for later processing
-            event_actions = _build_event_actions()
-            if event_actions:
-                setattr(func, EVENT_ACTIONS_MARKER, event_actions)
-            return func  # pyright: ignore [reportReturnType]
+            # A method: bound to its state class by __set_name__ when the
+            # class body is executed.
+            return EventHandler(fn=func, event_actions=event_actions)  # pyright: ignore [reportReturnType]
 
         if func is not None:
             return wrapper(func)
-        return wrapper
+        return wrapper  # pyright: ignore [reportReturnType]
 
     get_event = staticmethod(get_event)
     get_hydrate_event = staticmethod(get_hydrate_event)
diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py
index 66a88ac3e27..af4585b3d83 100644
--- a/packages/reflex-base/src/reflex_base/event/context.py
+++ b/packages/reflex-base/src/reflex_base/event/context.py
@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+import asyncio
 import dataclasses
 import functools
+import logging
 import os
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Protocol
 
 from reflex_base import otel
@@ -16,7 +18,12 @@
     from opentelemetry.context import Context
 
     from reflex.istate.manager import StateManager
+    from reflex.state import BaseState
     from reflex_base.event import Event
+    from reflex_base.event.processor.future import EventFuture
+
+
+logger = logging.getLogger(__name__)
 
 
 @functools.lru_cache
@@ -43,12 +50,15 @@ def _new_txid() -> str:
 class EnqueueProtocol(Protocol):
     """Protocol for the enqueue function in the event context."""
 
-    async def __call__(self, token: str, *events: Event) -> Any:
+    async def __call__(self, token: str, *events: Event) -> Sequence[EventFuture]:
         """Enqueue an event handler to be executed.
 
         Args:
             token: The client token associated with the event.
             events: The events to enqueue.
+
+        Returns:
+            One future per event, each resolving to its handler's return value.
         """
         ...
 
@@ -109,8 +119,19 @@ class EventContext(BaseContext):
     emit_event_impl: EmitEventProtocol | None = dataclasses.field(
         default=None, repr=False
     )
-    cached_states: dict[type, Any] = dataclasses.field(
-        default_factory=dict, init=False, repr=False
+    # The root of the state tree this task holds the token's state lock on,
+    # while it does: set by the event processor around a foreground handler,
+    # by ``async with self`` in a background handler, and by
+    # ``App.modify_state``. An awaited handler runs inline on it; with no lock
+    # held, the handler is enqueued instead. Never inherited by fork().
+    locked_root: BaseState | None = dataclasses.field(
+        default=None, init=False, repr=False
+    )
+    # The tasks running an awaited handler inline on the locked tree. The
+    # lock is released only once they are done, so a task spawned from a
+    # handler cannot keep writing to the tree after the lock is gone.
+    inline_tasks: set[asyncio.Task] = dataclasses.field(
+        default_factory=set, init=False, repr=False
     )
     # OpenTelemetry context active when this event was enqueued (None when tracing is off).
     otel_context: Context | None = dataclasses.field(default=None, repr=False)
@@ -143,6 +164,59 @@ def fork(
             otel_context=otel.capture_context(),
         )
 
+    def set_locked_root(self, root_state: BaseState | None) -> None:
+        """Record the state tree the current task holds the lock on, or None.
+
+        Args:
+            root_state: The locked root state, or None once the lock is released.
+        """
+        object.__setattr__(self, "locked_root", root_state)
+
+    async def release_lock(self) -> None:
+        """Wait for the inline handler runs started under the lock, then unmark it.
+
+        A handler run that another task started inline (``asyncio.ensure_future``
+        on a handler) still holds references to the locked tree, so the lock
+        stays held until it finishes. When the releasing task is being
+        cancelled, those runs are cancelled too.
+        """
+        if not self.inline_tasks:
+            # The common case: nothing was started inline.
+            self.set_locked_root(None)
+            return
+        current = asyncio.current_task()
+        # A task already being cancelled (the cancellation arrived inside
+        # the locked block) is not interrupted again while it waits, so the
+        # runs it waits for are cancelled up front.
+        cancelling = current is not None and current.cancelling()
+        try:
+            # Drain until nothing is left: a run may start another while
+            # the earlier ones are awaited.
+            while pending := [
+                t for t in self.inline_tasks if t is not current and not t.done()
+            ]:
+                if cancelling:
+                    for task in pending:
+                        task.cancel()
+                try:
+                    results = await asyncio.gather(*pending, return_exceptions=True)
+                except asyncio.CancelledError:
+                    for task in pending:
+                        task.cancel()
+                    raise
+                for task, result in zip(pending, results, strict=True):
+                    if isinstance(result, BaseException) and not isinstance(
+                        result, asyncio.CancelledError
+                    ):
+                        # Gathering retrieved the exception, so asyncio would
+                        # not report it as unretrieved; keep it visible.
+                        logger.error(
+                            f"Inline event handler task {task.get_name()} failed:",
+                            exc_info=result,
+                        )
+        finally:
+            self.set_locked_root(None)
+
     async def emit_delta(self, delta: Mapping[str, Mapping[str, Any]]) -> None:
         """Emit a delta to the frontend.
 
@@ -163,10 +237,13 @@ async def emit_event(self, *events: Event) -> None:
         if self.emit_event_impl is not None:
             await self.emit_event_impl(self.token, *events)
 
-    async def enqueue(self, *event: Event) -> None:
+    async def enqueue(self, *event: Event) -> Sequence[EventFuture]:
         """Enqueue an event handler to be executed.
 
         Args:
             event: The event to enqueue.
+
+        Returns:
+            One future per event, each resolving to its handler's return value.
         """
-        await self.enqueue_impl(self.token, *event)
+        return await self.enqueue_impl(self.token, *event)
diff --git a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py
index 7d13b83e437..637ed4c1cc9 100644
--- a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py
+++ b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import dataclasses
 import functools
 import inspect
@@ -66,6 +67,60 @@ def _connect_supersedes() -> tuple[str, str]:
     )
 
 
+_EVENT_LIKE_TYPES = (Event, EventHandler, EventSpec)
+
+
+def _is_event_like(value: Any) -> bool:
+    """Whether a handler's return value is an event or a sequence of events.
+
+    Args:
+        value: The value a handler returned or yielded.
+
+    Returns:
+        True for an Event, EventHandler or EventSpec, or a sequence of them.
+    """
+    if isinstance(value, _EVENT_LIKE_TYPES):
+        return True
+    if not is_non_string_sequence(value):
+        return False
+    try:
+        return all(isinstance(e, _EVENT_LIKE_TYPES) for e in value)
+    except TypeError:
+        return False
+
+
+def _is_empty_sequence(value: Any) -> bool:
+    """Whether a handler returned an empty sequence.
+
+    Args:
+        value: The value a handler returned.
+
+    Returns:
+        True for an empty non-string sequence.
+    """
+    return is_non_string_sequence(value) and len(value) == 0
+
+
+def _holds_events_or_coroutines(value: Any) -> bool:
+    """Whether a sequence holds an event or an unawaited coroutine.
+
+    Args:
+        value: The value a handler returned.
+
+    Returns:
+        True for a sequence ``_is_event_like`` rejects that still holds an
+        event (mixed with other values) or a coroutine.
+    """
+    if not is_non_string_sequence(value):
+        return False
+    try:
+        return any(
+            isinstance(e, _EVENT_LIKE_TYPES) or inspect.iscoroutine(e) for e in value
+        )
+    except TypeError:
+        return False
+
+
 def _check_valid_yield(events: Any, handler_name: str = "unknown") -> Any:
     """Check if the events yielded are valid. They must be EventHandlers or EventSpecs.
 
@@ -79,22 +134,12 @@ def _check_valid_yield(events: Any, handler_name: str = "unknown") -> Any:
     Raises:
         TypeError: If any of the events are not valid.
     """
-
-    def _is_valid_type(events: Any) -> bool:
-        return isinstance(events, (Event, EventHandler, EventSpec))
-
-    if events is None or _is_valid_type(events):
+    if events is None or _is_event_like(events):
         return events
 
     if not is_non_string_sequence(events):
         events = [events]
 
-    try:
-        if all(_is_valid_type(e) for e in events):
-            return events
-    except TypeError:
-        pass
-
     coroutines = [e for e in events if inspect.iscoroutine(e)]
 
     for coroutine in coroutines:
@@ -216,6 +261,54 @@ def _prepare_event_payload(
         return {arg: form_data_as_dict(value) for arg, value in payload.items()}
 
 
+def _written_snapshot(root_state: BaseState) -> dict[tuple[str, str], Any]:
+    """Capture the value of every dirty field in the tree, by identity.
+
+    Args:
+        root_state: The root of the state tree.
+
+    Returns:
+        The raw value of each dirty field of the root and its dirty substates,
+        keyed by state and field name, plus each state's in-place change count
+        under an empty field name. Comparing two snapshots with
+        ``_written_since`` tells whether a field was assigned or changed in
+        place in between.
+    """
+    snapshot: dict[tuple[str, str], Any] = {}
+    stack = [root_state]
+    while stack:
+        state = stack.pop()
+        cls = type(state)
+        fields = cls.__fields__
+        full_name = cls.get_full_name()
+        snapshot[full_name, ""] = state._mutations
+        for name in state.dirty_vars:
+            if (field := fields.get(name)) is not None:
+                snapshot[full_name, name] = field._get_raw(state)
+        substates = state.substates
+        stack.extend(substates[n] for n in state.dirty_substates if n in substates)
+    return snapshot
+
+
+def _written_since(
+    before: dict[tuple[str, str], Any], after: dict[tuple[str, str], Any]
+) -> bool:
+    """Whether a field was assigned between two snapshots.
+
+    Args:
+        before: The earlier snapshot.
+        after: The later snapshot.
+
+    Returns:
+        True if a field became dirty, holds a different object than before,
+        or a state's in-place change count moved (its own object, so unequal
+        counts are different objects too).
+    """
+    return before.keys() != after.keys() or any(
+        before[key] is not value for key, value in after.items()
+    )
+
+
 async def _route_events(ctx: EventContext, events: Sequence[Event]) -> None:
     """Emit frontend events to the client and queue backend events.
 
@@ -266,11 +359,27 @@ async def chain_updates(
         # machinery, so cleaning earlier would fan out a stale set (see
         # tests/integration/test_linked_state.py).
         try:
+            # Resolving suspends on async computed vars. While inline runs
+            # exist (this one included), a gathered sibling can assign a
+            # field meanwhile, after the delta was snapshotted; the clean
+            # below would drop that value unsent, so pick late assignments
+            # up until none happened during the resolve. A plain event has
+            # no inline runs and skips this. The clean runs before the emit:
+            # a write during the emit's await re-dirties its field instead
+            # of being cleaned, and the next flush sends it.
+            concurrent = bool(ctx.inline_tasks)
+            seen = _written_snapshot(root_state) if concurrent else {}
             delta = await root_state._get_resolved_delta()
-            if delta:
-                await ctx.emit_delta(delta)
+            while concurrent and _written_since(
+                seen, written := _written_snapshot(root_state)
+            ):
+                seen = written
+                for name, late in (await root_state._get_resolved_delta()).items():
+                    delta[name] = {**delta.get(name, {}), **late}
         finally:
             root_state._clean()
+        if delta:
+            await ctx.emit_delta(delta)
 
     # Convert valid EventHandler and EventSpec into Event
     if events is not None and (
@@ -282,7 +391,7 @@ async def chain_updates(
 
 
 def ensure_locked(
-    state: BaseState | StateProxy, root_state: BaseState | None
+    state: BaseState | StateProxy | None, root_state: BaseState | None
 ) -> BaseState | None:
     """The root to flush deltas from, only while the state lock is held.
 
@@ -308,18 +417,111 @@ def ensure_locked(
     return None
 
 
+async def execute_event_inline(spec: EventSpec) -> Any:
+    """Run the event a spec describes from inside the active EventContext.
+
+    This is what awaiting an EventHandler or EventSpec does. While the current
+    task holds the token's state lock (a foreground handler, ``async with
+    self`` in a background handler, ``app.modify_state``), the handler runs
+    inline on the locked state tree under the caller's context: its yields
+    emit deltas and the events it yields chain under the caller's event. With
+    no lock held, the event is enqueued and its future awaited. A stateless
+    handler always runs inline.
+
+    Args:
+        spec: The event spec to execute.
+
+    Returns:
+        The handler's return value.
+
+    Raises:
+        EventHandlerTypeError: If there is no active EventContext, the spec is
+            a frontend-only or upload event, or a background handler is
+            awaited while the state lock is held (which would deadlock).
+    """
+    from reflex_base.registry import RegistrationContext
+    from reflex_base.utils.exceptions import EventHandlerTypeError
+
+    try:
+        ctx = EventContext.get()
+    except LookupError as le:
+        msg = (
+            f"Event handler `{format_event_handler(spec.handler)}` can only be "
+            "awaited from an event handler or inside `app.modify_state`; use it "
+            "as an event trigger or yield it instead."
+        )
+        raise EventHandlerTypeError(msg) from le
+    event = spec._to_event(ctx.router_data)
+    name = event.name
+    if spec.client_handler_name:
+        msg = f"Upload event `{name}` cannot be awaited; use it as an event trigger."
+        raise EventHandlerTypeError(msg)
+    registered = RegistrationContext.get().event_handlers.get(name)
+    if registered is None:
+        msg = (
+            f"`{name}` is a frontend-only event, so it cannot be awaited; "
+            "yield or return it instead."
+        )
+        raise EventHandlerTypeError(msg)
+    handler = registered.handler
+    root_state = ctx.locked_root
+    if not registered.states:
+        if root_state is None:
+            return await process_event(
+                handler=handler, payload=event.payload, state=None, root_state=None
+            )
+    elif root_state is None:
+        (future,) = await ctx.enqueue(event)
+        return await future
+    elif handler.is_background:
+        msg = (
+            f"Background event handler `{name}` cannot be awaited while the "
+            "state lock is held, as it would wait for that lock; yield it instead."
+        )
+        raise EventHandlerTypeError(msg)
+    # Under the lock, a stateless handler flushes the caller's pending delta
+    # through the locked root before any frontend event it yields, like every
+    # other yield does, so it is covered by the lock like a state handler.
+    # The lock stays held until this run is done, even if the task that
+    # started it finishes first. Registered before the first await, so a
+    # task that starts while the lock is held is always seen by its release.
+    # A nested inline run on an already registered task leaves the outer
+    # run's registration in place.
+    task = asyncio.current_task()
+    if task is None or task in ctx.inline_tasks:
+        # Nothing to register here, and nothing to unregister after.
+        task = None
+    else:
+        ctx.inline_tasks.add(task)
+    try:
+        substate = (
+            await root_state.get_state(registered.states[0])
+            if registered.states
+            else None
+        )
+        return await process_event(
+            handler=handler,
+            payload=event.payload,
+            state=substate,
+            root_state=root_state,
+        )
+    finally:
+        if task is not None:
+            ctx.inline_tasks.discard(task)
+
+
 async def process_event(
     handler: EventHandler,
     payload: dict,
-    state: BaseState | StateProxy,
+    state: BaseState | StateProxy | None,
     root_state: BaseState | None,
-):
+) -> Any:
     """Process event.
 
     Args:
         handler: EventHandler to process.
         payload: The event payload.
-        state: State to process the handler.
+        state: State to process the handler, or None for a stateless handler.
         root_state: The root state of the app, used for emitting deltas. Pass
             None when the caller does not hold the state lock (background
             tasks): computing and cleaning a delta on an unlocked root races
@@ -327,6 +529,10 @@ async def process_event(
             changes are emitted by the ``async with self`` context exits
             instead.
 
+    Returns:
+        The handler's return value, None when it returned nothing or events
+        (which are chained instead).
+
     Raises:
         ValueError: If a string value is received for an int or float type and cannot be converted.
     """
@@ -334,13 +540,15 @@ async def process_event(
     handler_name = fn.__qualname__
     payload = _prepare_event_payload(handler, payload)
 
+    # A stateless handler takes no state argument.
+    args = () if state is None else (state,)
     # Handle async functions.
     if handler._is_coroutine_function:
-        events = await fn(state, **payload)
+        events = await fn(*args, **payload)
 
     # Handle regular functions.
     else:
-        events = fn(state, **payload)
+        events = fn(*args, **payload)
     # Handle async generators.
     if inspect.isasyncgen(events):
         async for event in events:
@@ -349,9 +557,7 @@ async def process_event(
                 root_state=ensure_locked(state, root_state),
                 handler_name=handler_name,
             )
-        await chain_updates(
-            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
-        )
+        events = None
 
     # Handle regular generators.
     elif inspect.isgenerator(events):
@@ -365,23 +571,28 @@ async def process_event(
         except StopIteration as si:
             # the "return" value of the generator is not available
             # in the loop, we must catch StopIteration to access it
-            if si.value is not None:
-                await chain_updates(
-                    si.value,
-                    root_state=ensure_locked(state, root_state),
-                    handler_name=handler_name,
-                )
-        await chain_updates(
-            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
-        )
+            events = si.value
 
-    # Handle regular event chains.
-    else:
+    # Chain a returned event (or events); anything else, an empty sequence
+    # included (a result such as no search hits, not an empty chain), is the
+    # result.
+    if events is None or (_is_event_like(events) and not _is_empty_sequence(events)):
         await chain_updates(
             events,
             root_state=ensure_locked(state, root_state),
             handler_name=handler_name,
         )
+        return None
+    # A coroutine that was never awaited, or events mixed with other values,
+    # is a bug, not a result: chain_updates flushes the delta and then raises
+    # for it, as it does for any invalid yield.
+    invalid = inspect.iscoroutine(events) or _holds_events_or_coroutines(events)
+    await chain_updates(
+        events if invalid else None,
+        root_state=ensure_locked(state, root_state),
+        handler_name=handler_name,
+    )
+    return events
 
 
 class BaseStateEventProcessor(EventProcessor):
@@ -528,9 +739,23 @@ def _on_disconnect(self, token: str) -> None:
         for future in self._superseded.pop((on_load_name, token), {}).values():
             future.cancel()
 
+    def _runs_unordered(self, registered_handler: RegisteredEventHandler) -> bool:
+        """Whether an event runs immediately instead of in its token's FIFO.
+
+        The FIFO serializes the handlers that take the token's state lock. A
+        stateless handler takes none, so it runs like a background one.
+
+        Args:
+            registered_handler: The registered handler for the event.
+
+        Returns:
+            True for a background or stateless handler.
+        """
+        return registered_handler.handler.is_background or not registered_handler.states
+
     async def _execute_event(
         self, *, entry: EventQueueEntry, registered_handler: RegisteredEventHandler
-    ) -> None:
+    ) -> Any:
         """Execute the handler for a single event queue entry with full state management.
 
         The ``EventContext`` has already been set by ``_process_event_queue_entry``
@@ -539,9 +764,20 @@ async def _execute_event(
         Args:
             entry: The event queue entry to process.
             registered_handler: The registered handler for the event.
+
+        Returns:
+            The handler's return value, which resolves the event's future.
         """
         ctx = entry.ctx
         event = entry.event
+        if not registered_handler.states:
+            # A stateless handler holds no lock and gets no state.
+            return await process_event(
+                handler=registered_handler.handler,
+                payload=event.payload,
+                state=None,
+                root_state=None,
+            )
         # The context, not the event: a chained event carries none of its own
         # and inherits the producing view's through fork().
         router_data = ctx.router_data
@@ -586,24 +822,29 @@ async def _execute_event(
                     await ctx.emit_delta(update.delta)
                 if update.events:
                     await _route_events(ctx, update.events)
-                return
+                return None
 
             # Get the event's substate.
             substate = await state.get_state(registered_handler.states[0])
             root_state = state._get_root_state()
 
-            if needs_to_rehydrate:
-                await self._rehydrate(root_state)
-
-            # Process non-background events while holding the lock.
-            if not registered_handler.handler.is_background:
-                await process_event(
-                    handler=registered_handler.handler,
-                    payload=event.payload,
-                    state=substate,
-                    root_state=root_state,
-                )
-                return
+            # Not `holding_lock`: this is the hot path of every event, and
+            # the plain try/finally spares the context manager's overhead.
+            ctx.set_locked_root(root_state)
+            try:
+                if needs_to_rehydrate:
+                    await self._rehydrate(root_state)
+
+                # Process non-background events while holding the lock.
+                if not registered_handler.handler.is_background:
+                    return await process_event(
+                        handler=registered_handler.handler,
+                        payload=event.payload,
+                        state=substate,
+                        root_state=root_state,
+                    )
+            finally:
+                await ctx.release_lock()
         # Otherwise drop the state lock and start processing the background task
         # with a proxy state. No root_state: the lock is no longer held, and
         # under a shared state tree (opportunistic locking, in-memory manager)
@@ -618,8 +859,9 @@ async def _execute_event(
         # lock, so these locals would pin an extra copy for the whole task.
         del state, substate, root_state
         handler_error: BaseException | None = None
+        result = None
         try:
-            await process_event(
+            result = await process_event(
                 handler=registered_handler.handler,
                 state=proxy,
                 payload=event.payload,
@@ -659,6 +901,7 @@ async def _execute_event(
                         "Error flushing delta after background handler "
                         f"{registered_handler.handler.fn.__qualname__} raised:"
                     )
+        return result
 
     async def _handle_backend_exception(
         self, ex: Exception, ev_ctx: EventContext | None = None
@@ -683,4 +926,9 @@ async def _handle_backend_exception(
                 )
 
 
-__all__ = ["BaseStateEventProcessor", "chain_updates", "process_event"]
+__all__ = [
+    "BaseStateEventProcessor",
+    "chain_updates",
+    "execute_event_inline",
+    "process_event",
+]
diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
index 3b76861899d..8bbbf0a4c38 100644
--- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
+++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py
@@ -19,6 +19,7 @@
 from reflex.istate.manager import StateManager
 from reflex_base import constants, otel
 from reflex_base.environment import environment
+from reflex_base.event import EventHandler, EventSpec
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor.blocking import (
     EVENT_TASK_PREFIX,
@@ -32,7 +33,7 @@
 
 if TYPE_CHECKING:
     from reflex.app import EventNamespace
-    from reflex.event import Event, EventSpec
+    from reflex.event import Event
 
 if hasattr(asyncio, "QueueShutDown"):
 
@@ -723,9 +724,20 @@ def _on_future_done(self, future: EventFuture) -> None:  # type: ignore[override
         if task is not None:
             task.cancel()
 
+    def _runs_unordered(self, registered_handler: RegisteredEventHandler) -> bool:
+        """Whether an event runs immediately instead of in its token's FIFO.
+
+        Args:
+            registered_handler: The registered handler for the event.
+
+        Returns:
+            True for a background handler.
+        """
+        return registered_handler.handler.is_background
+
     async def _execute_event(
         self, *, entry: EventQueueEntry, registered_handler: RegisteredEventHandler
-    ) -> None:
+    ) -> Any:
         """Execute the handler for a single event queue entry.
 
         This method contains the actual event-processing logic.  The base
@@ -740,15 +752,23 @@ async def _execute_event(
         Args:
             entry: The event queue entry to process.
             registered_handler: The registered handler for the event.
+
+        Returns:
+            The handler's return value, which resolves the event's future.
         """
         event = entry.event
         result = registered_handler.handler.fn(**event.payload)
-        if inspect.isawaitable(result):
-            await result
+        # An awaitable EventSpec or EventHandler the handler returns is a
+        # description of an event, not work to do here.
+        if inspect.isawaitable(result) and not isinstance(
+            result, (EventSpec, EventHandler)
+        ):
+            result = await result
+        return result
 
     async def _process_event_queue_entry(
         self, *, entry: EventQueueEntry, registered_handler: RegisteredEventHandler
-    ) -> None:
+    ) -> Any:
         """Process a single event queue entry.
 
         This function runs in a new task for each event.  It sets up the
@@ -763,16 +783,18 @@ async def _process_event_queue_entry(
         Args:
             entry: The event queue entry to process.
             registered_handler: The registered handler for the event.
+
+        Returns:
+            The handler's return value, which resolves the event's future.
         """
         # Set up the event context for this task.
         EventContext.set(entry.ctx)
         if not otel.enabled:
-            await self._execute_event(
+            return await self._execute_event(
                 entry=entry, registered_handler=registered_handler
             )
-            return
         with otel.event_span(entry.event, entry.ctx, registered_handler):
-            await self._execute_event(
+            return await self._execute_event(
                 entry=entry, registered_handler=registered_handler
             )
 
@@ -890,8 +912,7 @@ async def _process_queue(self):
                             f"No registered handler found for event: {entry.event.name}"
                         )
                         raise KeyError(msg) from ke
-                    if registered_handler.handler.is_background:
-                        # Background events run immediately, bypassing per-token ordering.
+                    if self._runs_unordered(registered_handler):
                         self._create_event_task(
                             entry=entry, registered_handler=registered_handler
                         )
diff --git a/packages/reflex-base/src/reflex_base/vars/base.py b/packages/reflex-base/src/reflex_base/vars/base.py
index 500579629ce..e2357971030 100644
--- a/packages/reflex-base/src/reflex_base/vars/base.py
+++ b/packages/reflex-base/src/reflex_base/vars/base.py
@@ -155,8 +155,10 @@ def _literal_var_for(value: Any) -> type[LiteralVar] | None:
         ),
         None,
     )
-    # A class object's type is its metaclass, which other classes share.
-    if not isinstance(value, type):
+    # A class object's type is its metaclass, which other classes share, and a
+    # proxy (such as a MutableProxy) reports the wrapped value's __class__, so
+    # its type says nothing about which literal class claims it.
+    if not isinstance(value, type) and value.__class__ is value_type:
         _literal_var_by_type[value_type] = literal_subclass
     return literal_subclass
 
@@ -4701,7 +4703,12 @@ def _validate_state_name(
     members = _reserved_state_members(root)
     if name not in members:
         return
+    from reflex_base.event import EventHandler
+
     method = value.__func__ if isinstance(value, (classmethod, staticmethod)) else value
+    if isinstance(method, EventHandler):
+        # A decorated method shadows like the function it wraps.
+        method = method.fn
     if isinstance(method, FunctionType):
         if value is members[name] or getattr(method, "__override_base_method__", False):
             return
diff --git a/reflex/app.py b/reflex/app.py
index 8237ea8adbe..7094d726091 100644
--- a/reflex/app.py
+++ b/reflex/app.py
@@ -2047,7 +2047,7 @@ class cannot be found.
             try:
                 forked_context = EventContext.get().fork(token=token.ident)
             except LookupError:
-                pass
+                forked_context = None
             else:
                 reset_token = EventContext.set(forked_context)
                 rebind.callback(EventContext.reset, reset_token)
@@ -2056,7 +2056,16 @@ class cannot be found.
                 token, previous_dirty_vars=previous_dirty_vars, **context
             ) as state:
                 # No other event handler can modify the state while in this context.
-                yield state
+                if forked_context is None:
+                    yield state
+                else:
+                    # An awaited handler runs inline on the locked tree. A plain
+                    # try/finally: this is also the shared-state fan-out path.
+                    forked_context.set_locked_root(state._get_root_state())
+                    try:
+                        yield state
+                    finally:
+                        await forked_context.release_lock()
                 delta = await state._get_resolved_delta()
                 state._clean()
                 if delta:
diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py
index a4df7c2401c..3c1b61f1ad0 100644
--- a/reflex/istate/proxy.py
+++ b/reflex/istate/proxy.py
@@ -286,6 +286,8 @@ async def __aenter__(self) -> Self:
             super().__setattr__(
                 "__wrapped__", mutable_state.get_substate(self._self_substate_path)
             )
+            # An awaited handler runs inline on this tree while the lock is held.
+            ctx.set_locked_root(mutable_state)
             from reflex.state import State
 
             if type(mutable_state) is State and not mutable_state.router_data:
@@ -325,7 +327,12 @@ async def __aexit__(self, *exc_info: Any) -> None:
         finally:
             try:
                 if self._self_mutable and self._self_actx is not None:
-                    await self._self_actx.__aexit__(*exc_info)
+                    try:
+                        await EventContext.get().release_lock()
+                    finally:
+                        # Waiting for inline runs may be cancelled; the state
+                        # manager's lock must be released regardless.
+                        await self._self_actx.__aexit__(*exc_info)
             finally:
                 self._self_actx = None
                 self._self_mutable = False
@@ -946,6 +953,7 @@ def _mark_dirty(
         """
         state = self._dirty_state()
         type(state).__fields__[self._self_field_name]._mark_dirty(state)
+        state._note_in_place_change()
         if wrapped is not None:
             return wrapped(*args, **(kwargs or {}))
         return None
diff --git a/reflex/state.py b/reflex/state.py
index b4e20f2257b..e10c6856e49 100644
--- a/reflex/state.py
+++ b/reflex/state.py
@@ -35,7 +35,9 @@
 from reflex_base.constants.state import FIELD_MARKER
 from reflex_base.environment import PerformanceMode, auto_reload_cooldown, environment
 from reflex_base.event import (
+    BACKGROUND_TASK_MARKER,
     EVENT_ACTIONS_MARKER,
+    SUPERSEDES_MARKER,
     Event,
     EventHandler,
     EventSpec,
@@ -543,6 +545,22 @@ def __get__(self, instance: Any, owner: type | None = None, /):
 _ROUTER_FIELD_NAMES = frozenset((*constants.ROUTER_VARS, constants.ROUTER_DATA))
 
 
+# What a wrapper stacked above @rx.event takes over from the function it
+# wraps: what functools.wraps copied from a plain function (its name, module,
+# docstring and the annotations the payload coercion reads, which it took
+# from the EventHandler class instead) and the markers @rx.event set on it.
+_STACKED_HANDLER_ATTRIBUTES = (
+    "__module__",
+    "__name__",
+    "__qualname__",
+    "__doc__",
+    "__annotations__",
+    BACKGROUND_TASK_MARKER,
+    SUPERSEDES_MARKER,
+    EVENT_ACTIONS_MARKER,
+)
+
+
 class BaseState(EvenMoreBasicBaseState, state_root=True):
     """The state of the app."""
 
@@ -585,6 +603,7 @@ class BaseState(EvenMoreBasicBaseState, state_root=True):
 
     # Instance bookkeeping, kept out of `__dict__`: never a field, never pickled.
     __slots__ = (
+        "_mutations",
         "_was_touched",
         "dirty_substates",
         "dirty_vars",
@@ -604,6 +623,9 @@ class BaseState(EvenMoreBasicBaseState, state_root=True):
         dirty_substates: set[str] = field(default_factory=set, is_var=False)
         # Whether the state was modified since it was last persisted.
         _was_touched: bool = field(default=False, is_var=False)
+        # How many in-place changes its mutable values took; a flush that
+        # suspended compares it, since such a change keeps the value's identity.
+        _mutations: int = field(default=0, is_var=False)
 
     # The routing path that triggered the state
     router_data: builtins.dict[str, Any] = field(
@@ -699,6 +721,7 @@ def _init_bookkeeping(self, parent_state: BaseState | None = None) -> None:
         setattr_(self, "dirty_vars", set())
         setattr_(self, "dirty_substates", set())
         setattr_(self, "_was_touched", False)
+        setattr_(self, "_mutations", 0)
 
     def __repr__(self) -> str:
         """Get the string representation of the state.
@@ -803,7 +826,12 @@ def __init_subclass__(cls, mixin: bool = False, **kwargs):
 
         # Set up the event handlers.
         for name, fn in list(cls.__dict__.items()):
-            if cls._item_is_event_handler(name, fn):
+            if isinstance(fn, EventHandler):
+                # Bound by __set_name__ (a decorated method). One bound to
+                # another state is an alias and stays that state's.
+                if fn.state is cls and name not in cls.event_handlers:
+                    cls.event_handlers[name] = fn
+            elif cls._item_is_event_handler(name, fn):
                 handler = cls._create_event_handler(fn)
                 cls.event_handlers[name] = handler
                 setattr(cls, name, handler)
@@ -864,6 +892,12 @@ def _bind_mixin_members(cls) -> None:
                             _var_data=VarData.from_state(cls),
                         ),
                     )
+                elif isinstance(value, EventHandler) and value.state is mixin_cls:
+                    # A handler declared on the mixin: copy its function, which
+                    # carries the markers, and let it be wrapped for this class.
+                    fn = cls._copy_fn(value.fn)
+                    fn.__qualname__ = f"{cls.__name__}.{name}"
+                    setattr(cls, name, fn)
                 elif cls._item_is_event_handler(name, value):
                     fn = cls._copy_fn(value)
                     fn.__qualname__ = f"{cls.__name__}.{name}"
@@ -918,7 +952,28 @@ def _item_is_event_handler(name: str, value: Any) -> bool:
 
         Returns:
             Whether the item is an event handler.
+
+        A decorator stacked above ``@rx.event`` (found through the whole
+        ``__wrapped__`` chain) is deprecated: the wrapper is treated as the
+        plain method it wraps, as when ``@rx.event`` returned the function,
+        carrying the inner handler's name and markers.
         """
+        if not isinstance(value, EventHandler) and isinstance(
+            inner := inspect.unwrap(value, stop=lambda f: isinstance(f, EventHandler)),
+            EventHandler,
+        ):
+            console.deprecate(
+                feature_name="Decorators above @rx.event",
+                reason=(
+                    f"`{name}` is decorated above @rx.event; apply other "
+                    "decorators below it."
+                ),
+                deprecation_version="0.10.2",
+                removal_version="1.0",
+            )
+            for attr in _STACKED_HANDLER_ATTRIBUTES:
+                if hasattr(inner.fn, attr):
+                    setattr(value, attr, getattr(inner.fn, attr))
         return (
             not name.startswith("_")
             and isinstance(value, Callable)
@@ -1958,6 +2013,15 @@ async def _get_resolved_delta(self) -> Delta:
             await _sync_shared_storage(self)
         return await resolve_delta(self)
 
+    def _note_in_place_change(self) -> None:
+        """Count an in-place change to one of this state's mutable values.
+
+        Called through a proxy, so the count lands on the state itself. Kept
+        off the plain assignment path, which is only a few hundred
+        nanoseconds and detects reassignment by identity instead.
+        """
+        object.__setattr__(self, "_mutations", self._mutations + 1)
+
     def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None:
         """Mark this state and its ancestors dirty, invalidating dependent computed vars.
 
diff --git a/tests/integration/test_background_task.py b/tests/integration/test_background_task.py
index 283dbe4cd12..45055f8d8a9 100644
--- a/tests/integration/test_background_task.py
+++ b/tests/integration/test_background_task.py
@@ -114,6 +114,25 @@ async def yield_in_async_with_self(self):
                 yield
                 self.counter += 1
 
+        @rx.event(background=True)
+        async def await_handlers(self):
+            # No lock held: each await enqueues the handler and waits for it.
+            await State.increment()
+            await State.increment()
+            async with self:
+                # Lock held: the handler runs inline on the locked state.
+                await State.increment()
+                assert self.counter == 3
+                await State.increment_by(await State.pick_amount(2))
+
+        @rx.event
+        def increment_by(self, amount: int):
+            self.counter += amount
+
+        @rx.event
+        def pick_amount(self, amount: int) -> int:
+            return amount * 2
+
         @rx.event(background=True)
         async def disconnect_reconnect_background(self):
             async with self:
@@ -212,6 +231,11 @@ def index() -> rx.Component:
                 on_click=State.fast_yielding,
                 id="fast-yielding",
             ),
+            rx.button(
+                "Await Handlers",
+                on_click=State.await_handlers,
+                id="await-handlers",
+            ),
             rx.button("Reset", on_click=State.reset_counter, id="reset"),
         )
 
@@ -424,6 +448,28 @@ def test_yield_in_async_with_self(
     AppHarness.expect(lambda: counter.text == "2", timeout=5)
 
 
+def test_await_handlers_from_background_task(
+    background_task: AppHarness,
+    driver: WebDriver,
+    token: str,
+):
+    """A background task awaits handlers through the queue, or inline under the lock.
+
+    Args:
+        background_task: harness for BackgroundTask app.
+        driver: WebDriver instance.
+        token: The token for the connected client.
+    """
+    assert background_task.app_instance is not None
+
+    await_handlers_button = driver.find_element(By.ID, "await-handlers")
+    counter = driver.find_element(By.ID, "counter")
+    AppHarness.expect(lambda: counter.text == "0", timeout=5)
+
+    await_handlers_button.click()
+    AppHarness.expect(lambda: counter.text == "7", timeout=5)
+
+
 @pytest.mark.parametrize(
     "button_id",
     [
diff --git a/tests/integration/test_event_chain.py b/tests/integration/test_event_chain.py
index 2059095bf6d..2b5f4fede41 100644
--- a/tests/integration/test_event_chain.py
+++ b/tests/integration/test_event_chain.py
@@ -116,6 +116,19 @@ def click_yield_nested(self):
             yield State.event_nested_1
             yield State.event_arg("yield_nested")
 
+        @rx.event
+        def event_return_value(self, arg: int):
+            self.event_order.append(f"event_return_value:{arg}")
+            yield State.event_arg(f"from_callee:{arg}")
+            return arg * 2  # noqa: B901
+
+        @rx.event
+        async def click_await_handler(self):
+            self.event_order.append("click_await_handler:0")
+            value = await State.event_return_value(21)
+            self.event_order.append(f"click_await_handler:{value}")
+            yield State.event_arg("after_await")
+
         @rx.event
         def redirect_return_chain(self):
             self.event_order.append("redirect_return_chain")
@@ -204,6 +217,11 @@ def index():
                 id="yield_nested",
                 on_click=State.click_yield_nested,
             ),
+            rx.button(
+                "Await Handler",
+                id="await_handler",
+                on_click=State.click_await_handler,
+            ),
             rx.button(
                 "Redirect Yield Chain",
                 id="redirect_yield_chain",
@@ -446,6 +464,16 @@ def assert_token(event_chain: AppHarness, driver: WebDriver) -> str:
                 "event_arg:nested_3",
             ],
         ),
+        (
+            "await_handler",
+            [
+                "click_await_handler:0",
+                "event_return_value:21",
+                "click_await_handler:42",
+                "event_arg:from_callee:21",
+                "event_arg:after_await",
+            ],
+        ),
         (
             "redirect_return_chain",
             [
diff --git a/tests/units/istate/test_proxy.py b/tests/units/istate/test_proxy.py
index 9aa5404bf5a..cb839002c34 100644
--- a/tests/units/istate/test_proxy.py
+++ b/tests/units/istate/test_proxy.py
@@ -1502,3 +1502,21 @@ async def test_immutable_mutable_proxy_async_context_dict_view_paths(
     ) as state:
         assert isinstance(state, NestedMutableProxyState)
         assert state.data == {"a": {"x": 10}, "b": {"y": 20}}
+
+
+@pytest.mark.asyncio
+async def test_proxy_context_sets_and_clears_locked_root(
+    attached_mock_event_context: EventContext,
+):
+    """``async with self`` marks the context's locked root for its duration.
+
+    Args:
+        attached_mock_event_context: The attached event context fixture.
+    """
+    state_proxy = StateProxy(ProxyTestState())
+    assert attached_mock_event_context.locked_root is None
+    async with state_proxy as mutable:
+        locked_root = attached_mock_event_context.locked_root
+        assert locked_root is not None
+        assert locked_root is mutable.__wrapped__._get_root_state()  # pyright: ignore[reportAttributeAccessIssue]
+    assert attached_mock_event_context.locked_root is None
diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py
index 7675cbd08d2..0ad05259b5b 100644
--- a/tests/units/reflex_base/event/processor/test_base_state_processor.py
+++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py
@@ -3503,3 +3503,1467 @@ def get_delta(self) -> Delta:
         state = await root.get_state(StorageState)
         assert state.value == "browser"
         assert state.revision == 1
+
+
+async def test_handler_return_value_resolves_future(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """A non-event return value resolves the event's future after the delta.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: List to capture emitted deltas.
+        token: The client token.
+    """
+
+    class ReturnState(State):
+        n: int = 0
+
+        @event
+        def compute(self, x: int) -> int:
+            self.n = x
+            return x * 2
+
+        @event
+        async def compute_async(self, x: int) -> dict[str, int]:
+            self.n = x
+            return {"doubled": x * 2}
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(ReturnState.compute(21))[0]
+        )
+        assert await future == 42
+        assert any(
+            delta.get(ReturnState.get_full_name(), {}).get(f"n{FIELD_MARKER}") == 21
+            for _, delta in emitted_deltas
+        )
+        future = await processor.enqueue(
+            token, Event.from_event_type(ReturnState.compute_async(4))[0]
+        )
+        assert await future == {"doubled": 8}
+
+
+async def test_generator_return_value_resolves_future(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """A generator handler's ``return`` value resolves the future; a returned event chains.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: List to capture emitted deltas.
+        token: The client token.
+    """
+
+    class GenReturnState(State):
+        n: int = 0
+        log: list[str] = []
+
+        @event
+        def steps(self):
+            self.n = 1
+            yield
+            self.n = 2
+            return "done"  # noqa: B901
+
+        @event
+        def add(self, tag: str):
+            self.log = [*self.log, tag]
+
+        @event
+        def chains(self):
+            yield
+            return GenReturnState.add("chained")  # noqa: B901
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(GenReturnState.steps())[0]
+        )
+        assert await future == "done"
+        n_values = [
+            delta[GenReturnState.get_full_name()].get(f"n{FIELD_MARKER}")
+            for _, delta in emitted_deltas
+            if GenReturnState.get_full_name() in delta
+        ]
+        # The hydration snapshot carries the default before the two yields.
+        assert n_values[-2:] == [1, 2]
+        future = await processor.enqueue(
+            token, Event.from_event_type(GenReturnState.chains())[0]
+        )
+        assert await future.wait_all() is None
+        await processor.join(5)
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(GenReturnState)).log == ["chained"]
+
+
+async def test_returned_coroutine_still_raises_type_error(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A handler returning a coroutine it forgot to await is an error, not a result.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    async def _work() -> int:  # noqa: RUF029
+        return 1
+
+    class ForgotAwaitState(State):
+        @event
+        def handler(self):
+            return _work()
+
+    captured: list[Exception] = []
+    real_base_state_processor.backend_exception_handler = captured.append
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(ForgotAwaitState.handler())[0]
+        )
+        with pytest.raises(TypeError, match="must only return/yield"):
+            await future
+    assert len(captured) == 1
+
+
+async def test_returned_list_mixing_events_and_values_raises_type_error(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A returned sequence that holds an event next to a plain value is still an error.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class MixedReturnState(State):
+        log: list[str] = []
+
+        @event
+        def add(self):
+            self.log = [*self.log, "add"]
+
+        @event
+        def handler(self):
+            return [MixedReturnState.add(), "x"]
+
+    captured: list[Exception] = []
+    real_base_state_processor.backend_exception_handler = captured.append
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(MixedReturnState.handler())[0]
+        )
+        with pytest.raises(TypeError, match="must only return/yield"):
+            await future
+    assert len(captured) == 1
+
+
+async def test_returned_list_of_coroutines_raises_type_error(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A returned list of coroutines that were never awaited is an error, not a result.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    async def _work() -> int:  # noqa: RUF029
+        return 1
+
+    class ForgotAwaitListState(State):
+        @event
+        def handler(self):
+            return [_work(), _work()]
+
+    captured: list[Exception] = []
+    real_base_state_processor.backend_exception_handler = captured.append
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(ForgotAwaitListState.handler())[0]
+        )
+        with pytest.raises(TypeError, match="must only return/yield"):
+            await future
+    assert len(captured) == 1
+
+
+async def test_invalid_return_value_still_flushes_the_delta_first(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list,
+    token: str,
+):
+    """A handler's state changes reach the frontend before an invalid return value raises.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: Deltas emitted to the frontend, in order.
+        token: The client token.
+    """
+
+    async def _work() -> int:  # noqa: RUF029
+        return 1
+
+    class FlushThenRaiseState(State):
+        value: str = "old"
+
+        @event
+        def handler(self):
+            self.value = "new"
+            return _work()
+
+    deltas_before_error: list[int] = []
+    real_base_state_processor.backend_exception_handler = lambda _ex: (
+        deltas_before_error.append(len(emitted_deltas))
+    )
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(FlushThenRaiseState.handler())[0]
+        )
+        with pytest.raises(TypeError, match="must only return/yield"):
+            await future
+    flushed = [
+        delta
+        for _, delta in emitted_deltas[: deltas_before_error[0]]
+        if delta.get(FlushThenRaiseState.get_full_name(), {}).get(
+            "value" + FIELD_MARKER
+        )
+        == "new"
+    ]
+    assert flushed, "the change was not emitted before the error was handled"
+
+
+async def test_handler_returning_an_empty_list_resolves_to_the_list(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """An empty list is a result, not an empty event chain, so awaiters get it back.
+
+    Yielding an empty list is still a no-op, as before.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class EmptyResultState(State):
+        @event
+        def search(self) -> list[str]:
+            return []
+
+        @event
+        def yields_nothing(self):
+            yield []
+
+    async with real_base_state_processor as processor:
+        assert (
+            await (
+                await processor.enqueue(
+                    token, Event.from_event_type(EmptyResultState.search())[0]
+                )
+            )
+            == []
+        )
+        assert (
+            await (
+                await processor.enqueue(
+                    token, Event.from_event_type(EmptyResultState.yields_nothing())[0]
+                )
+            )
+            is None
+        )
+
+
+async def test_background_handler_return_value_resolves_future(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A background handler's return value resolves its future too.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class BgReturnState(State):
+        n: int = 0
+
+        @event(background=True)
+        async def compute(self, x: int) -> int:
+            async with self:
+                self.n = x
+            return x + 1
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(BgReturnState.compute(1))[0]
+        )
+        assert await future == 2
+
+
+async def test_stateless_handler_runs_without_lock_and_resolves_result(
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A stateless handler runs with no state and its return value resolves the future.
+
+    Args:
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    calls: list[tuple[int, int]] = []
+
+    @event
+    def stateless_add(a: int, b: int) -> int:
+        calls.append((a, b))
+        return a + b
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(stateless_add(1, 2))[0]
+        )
+        assert await future == 3
+    assert calls == [(1, 2)]
+
+
+async def test_stateless_handler_chains_returned_events(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """Events a stateless handler returns or yields run against the token's state.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class ChainTargetState(State):
+        log: list[str] = []
+
+        @event
+        def add(self, tag: str):
+            self.log = [*self.log, tag]
+
+    @event
+    def stateless_chain():
+        yield ChainTargetState.add("yielded")
+        return ChainTargetState.add("returned")  # noqa: B901
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(stateless_chain())[0]
+        )
+        assert await future.wait_all() is None
+        await processor.join(5)
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(ChainTargetState)).log == ["yielded", "returned"]
+
+
+async def test_stateless_handler_is_not_serialized_behind_the_token_queue(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A stateless handler holds no lock, so it does not wait for the token's FIFO.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    order: list[str] = []
+    release = asyncio.Event()
+
+    class SlowState(State):
+        @event
+        async def slow(self):
+            await release.wait()
+            order.append("slow")
+
+    @event
+    def quick():
+        order.append("quick")
+
+    async with real_base_state_processor as processor:
+        slow_future = await processor.enqueue(
+            token, Event.from_event_type(SlowState.slow())[0]
+        )
+        quick_future = await processor.enqueue(token, Event.from_event_type(quick())[0])
+        await asyncio.wait_for(quick_future, 5)
+        assert order == ["quick"]
+        release.set()
+        await asyncio.wait_for(slow_future, 5)
+    assert order == ["quick", "slow"]
+
+
+async def test_locked_root_is_set_only_while_the_lock_is_held(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """The context names the locked root during a handler, and None otherwise.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    seen: dict[str, Any] = {}
+
+    class LockState(State):
+        @event
+        def foreground(self):
+            ctx = EventContext.get()
+            seen["fg"] = ctx.locked_root is self._get_root_state()
+            seen["fg_ctx"] = ctx
+
+        @event(background=True)
+        async def background(self):
+            ctx = EventContext.get()
+            seen["bg_outside"] = ctx.locked_root
+            async with self:
+                seen["bg_inside"] = (
+                    ctx.locked_root is self.__wrapped__._get_root_state()  # pyright: ignore[reportAttributeAccessIssue]
+                )
+            seen["bg_after"] = ctx.locked_root
+
+        @event
+        def raises(self):
+            seen["raises_ctx"] = EventContext.get()
+            msg = "boom"
+            raise ValueError(msg)
+
+    captured: list[Exception] = []
+    real_base_state_processor.backend_exception_handler = captured.append
+    async with real_base_state_processor as processor:
+        await processor.enqueue(token, Event.from_event_type(LockState.foreground())[0])
+        await processor.enqueue(token, Event.from_event_type(LockState.background())[0])
+        future = await processor.enqueue(
+            token, Event.from_event_type(LockState.raises())[0]
+        )
+        with pytest.raises(ValueError, match="boom"):
+            await future
+        await processor.join(5)
+        await asyncio.sleep(0)
+
+    assert seen["fg"] is True
+    assert seen["fg_ctx"].locked_root is None
+    assert seen["bg_outside"] is None
+    assert seen["bg_inside"] is True
+    assert seen["bg_after"] is None
+    assert seen["raises_ctx"].locked_root is None
+    assert len(captured) == 1
+
+
+async def test_await_handler_inline_returns_value_and_chains_its_yields(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """Awaiting a handler runs it under the caller's lock; its yields chain under the caller.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: List to capture emitted deltas.
+        token: The client token.
+    """
+
+    class AwaitState(State):
+        log: list[str] = []
+        n: int = 0
+
+        @event
+        def add(self, tag: str):
+            self.log = [*self.log, tag]
+
+        @event
+        def callee(self, x: int):
+            self.log = [*self.log, f"callee:{x}"]
+            self.n = x
+            yield AwaitState.add("from_callee")
+            return x * 2  # noqa: B901
+
+        @event
+        async def caller(self):
+            self.log = [*self.log, "caller:0"]
+            value = await AwaitState.callee(21)
+            self.log = [*self.log, f"caller:{value}"]
+            await asyncio.gather(AwaitState.add("g1"), AwaitState.add("g2"))
+            yield AwaitState.add("after_await")
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(AwaitState.caller())[0]
+        )
+        # The chained events are children of the caller's future, so waiting
+        # on the tree is enough: no processor.join needed.
+        await future.wait_all()
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(AwaitState)).log == [
+        "caller:0",
+        "callee:21",
+        "caller:42",
+        "g1",
+        "g2",
+        "from_callee",
+        "after_await",
+    ]
+    # The callee's yield flushed a delta with its change before the caller went on.
+    n_values = [
+        delta[AwaitState.get_full_name()][f"n{FIELD_MARKER}"]
+        for _, delta in emitted_deltas
+        if f"n{FIELD_MARKER}" in delta.get(AwaitState.get_full_name(), {})
+    ]
+    assert 21 in n_values
+
+
+async def test_await_from_substate_targets_sibling_state(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A handler can await a handler of another state in the same tree.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class SiblingA(State):
+        total: int = 0
+
+        @event
+        async def run(self):
+            self.total = await SiblingB.compute(5)
+
+    class SiblingB(State):
+        seen: int = 0
+
+        @event
+        def compute(self, x: int) -> int:
+            self.seen = x
+            return x + 1
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(SiblingA.run())[0]
+        )
+        await future
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(SiblingA)).total == 6
+    assert (await state.get_state(SiblingB)).seen == 5
+
+
+async def test_await_background_handler_from_locked_context_raises(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """Awaiting a background handler under the state lock is refused (it would deadlock).
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    from reflex_base.utils.exceptions import EventHandlerTypeError
+
+    class BgAwaitState(State):
+        @event(background=True)
+        async def work(self):
+            pass
+
+        @event
+        async def caller(self):
+            await BgAwaitState.work()
+
+    captured: list[Exception] = []
+    real_base_state_processor.backend_exception_handler = captured.append
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(BgAwaitState.caller())[0]
+        )
+        with pytest.raises(EventHandlerTypeError, match="yield it instead"):
+            await future
+
+
+async def test_await_from_background_enqueues_outside_and_runs_inline_inside(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A background handler awaits through the queue, or inline inside ``async with self``.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    results: list[int] = []
+
+    class BgCallerState(State):
+        counter: int = 0
+
+        @event
+        def increment(self) -> int:
+            self.counter += 1
+            return self.counter
+
+        @event(background=True)
+        async def run(self):
+            # No lock held: each await enqueues the event and waits for it.
+            results.append(await BgCallerState.increment())
+            results.append(await BgCallerState.increment())
+            async with self:
+                # Lock held: runs inline on the locked tree.
+                results.append(await BgCallerState.increment())
+                assert self.counter == 3
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(BgCallerState.run())[0]
+        )
+        await future.wait_all()
+
+    assert results == [1, 2, 3]
+
+
+async def test_await_without_event_context_raises():
+    """Awaiting a handler outside any EventContext is an error."""
+    from reflex_base.utils.exceptions import EventHandlerTypeError
+
+    class NoCtxState(State):
+        @event
+        def handle(self):
+            pass
+
+    with pytest.raises(EventHandlerTypeError, match="can only be awaited"):
+        await NoCtxState.handle()
+    with pytest.raises(EventHandlerTypeError, match="can only be awaited"):
+        await NoCtxState.handle
+
+
+async def test_await_frontend_only_and_upload_specs_raise(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """Frontend-only events and upload events cannot be awaited.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    from reflex_base.utils.exceptions import EventHandlerTypeError
+
+    errors: dict[str, str] = {}
+
+    class FrontendAwaitState(State):
+        @event
+        async def handle_upload(self, files: list[rx.UploadFile]):
+            pass
+
+        @event
+        async def caller(self):
+            try:
+                await rx.redirect("/")
+            except EventHandlerTypeError as ex:
+                errors["redirect"] = str(ex)
+            try:
+                await FrontendAwaitState.handle_upload(rx.upload_files(upload_id="u"))
+            except EventHandlerTypeError as ex:
+                errors["upload"] = str(ex)
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(FrontendAwaitState.caller())[0]
+        )
+        await future
+
+    assert "frontend-only" in errors["redirect"]
+    assert "cannot be awaited" in errors["upload"]
+
+
+async def test_await_same_spec_twice_executes_twice(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A spec describes an event, so each await runs it again.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class TwiceState(State):
+        log: list[str] = []
+
+        @event
+        def add(self, tag: str):
+            self.log = [*self.log, tag]
+
+        @event
+        async def caller(self):
+            spec = TwiceState.add("x")
+            await spec
+            await spec
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(TwiceState.caller())[0]
+        )
+        await future
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(TwiceState)).log == ["x", "x"]
+
+
+async def test_await_payload_decodes_vars_and_detaches_proxies(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """Args bound to an awaited handler arrive as plain values, detached from the state.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    received: list[Any] = []
+
+    class PayloadState(State):
+        items: list[int] = [1, 2]
+
+        @event
+        def take(self, values: list[int]) -> int:
+            received.append(values)
+            values.append(99)
+            return len(values)
+
+        @event(background=True)
+        async def run(self):
+            async with self:
+                assert await PayloadState.take(self.items) == 3
+                assert self.items == [1, 2]
+            assert await PayloadState.take(self.items) == 3
+
+    async with real_base_state_processor as processor:
+        future = await processor.enqueue(
+            token, Event.from_event_type(PayloadState.run())[0]
+        )
+        await future.wait_all()
+
+    assert [type(v) for v in received] == [list, list]
+    assert received == [[1, 2, 99], [1, 2, 99]]
+
+
+async def test_await_stateless_handler_flushes_the_caller_delta_first(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list,
+    emitted_events: list,
+    token: str,
+):
+    """A stateless helper awaited under the lock emits the caller's pending delta first.
+
+    A script the helper yields must see the state the caller just changed,
+    which is the ordering every other yield keeps.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: Deltas emitted to the frontend, in order.
+        emitted_events: Frontend events emitted, in order.
+        token: The client token.
+    """
+    deltas_before_script: list[int] = []
+
+    @event
+    def read_storage():
+        yield rx.call_script("localStorage.getItem('value')")
+        deltas_before_script.append(len(emitted_deltas))
+
+    class FlushFirstState(State):
+        value: str = "old"
+
+        @event
+        async def change_then_read(self):
+            self.value = "new"
+            await read_storage()
+
+    async with real_base_state_processor as processor:
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(FlushFirstState.change_then_read())[0]
+            )
+        )
+
+    assert len(emitted_events) == 1
+    assert "_call_script" in emitted_events[0][1][0].name
+    flushed_before_script = [
+        delta
+        for _, delta in emitted_deltas[: deltas_before_script[0]]
+        if delta.get(FlushFirstState.get_full_name(), {}).get("value" + FIELD_MARKER)
+        == "new"
+    ]
+    assert flushed_before_script, (
+        "the caller's change reached the frontend after the script"
+    )
+
+
+async def test_await_stateless_handler_runs_inline(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A stateless handler is awaited inline, with or without the lock held.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    @event
+    def stateless_double(x: int) -> int:
+        return x * 2
+
+    class StatelessCallerState(State):
+        got: int = 0
+
+        @event
+        async def caller(self):
+            self.got = await stateless_double(4)
+
+        @event(background=True)
+        async def bg_caller(self):
+            value = await stateless_double(5)
+            async with self:
+                self.got = value
+
+    async with real_base_state_processor as processor:
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(StatelessCallerState.caller())[0]
+            )
+        )
+        root_ctx = real_base_state_processor._root_context
+        assert root_ctx is not None
+        state = await root_ctx.state_manager.get_state(
+            BaseStateToken(ident=token, cls=State)
+        )
+        assert (await state.get_state(StatelessCallerState)).got == 8
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(StatelessCallerState.bg_caller())[0]
+            )
+        )
+        state = await root_ctx.state_manager.get_state(
+            BaseStateToken(ident=token, cls=State)
+        )
+        assert (await state.get_state(StatelessCallerState)).got == 10
+
+
+async def test_detached_stateless_helper_holds_the_lock_until_it_finishes(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A stateless helper started from a locked handler keeps the lock until it is done.
+
+    It flushes deltas through the caller's root, so the lock must outlive it
+    like it outlives a detached state handler.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    release = asyncio.Event()
+    seen: dict[str, Any] = {}
+
+    @event
+    async def helper():
+        await release.wait()
+        yield rx.call_script("void 0")
+
+    class DetachedHelperState(State):
+        log: list[str] = []
+
+        @event
+        async def parent(self):
+            seen["ctx"] = EventContext.get()
+            asyncio.ensure_future(helper())  # noqa: RUF006
+            # Let the helper start under the lock before returning.
+            await asyncio.sleep(0)
+            self.log = [*self.log, "parent"]
+
+        @event
+        def next_event(self):
+            self.log = [*self.log, "next"]
+
+    async with real_base_state_processor as processor:
+        parent_future = await processor.enqueue(
+            token, Event.from_event_type(DetachedHelperState.parent())[0]
+        )
+        next_future = await processor.enqueue(
+            token, Event.from_event_type(DetachedHelperState.next_event())[0]
+        )
+        await asyncio.sleep(0.05)
+        # The parent has returned, but its lock is still held for the helper.
+        assert not parent_future.done()
+        assert not next_future.done()
+        assert seen["ctx"].locked_root is not None
+        release.set()
+        await asyncio.wait_for(next_future, 5)
+        assert parent_future.done()
+
+    assert seen["ctx"].locked_root is None
+
+
+async def test_gathered_inline_handlers_keep_each_others_changes(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list,
+    token: str,
+):
+    """A change made while another gathered handler's delta resolves still reaches the frontend.
+
+    Resolving a delta suspends on an async computed var. A gathered handler
+    that changes state meanwhile, and only flushes after the first one has
+    cleaned, must not have its change cleaned away unsent.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: Deltas emitted to the frontend, in order.
+        token: The client token.
+    """
+    resolving = asyncio.Event()
+
+    class GatheredState(State):
+        a: int = 0
+        b: int = 0
+
+        @rx.var(cache=False)
+        async def slow(self) -> int:
+            # Resolving set_a's delta: let set_b write now, after the snapshot.
+            if self.a == 1:
+                resolving.set()
+            await asyncio.sleep(0)
+            return self.a
+
+        @event
+        async def set_a(self):
+            self.a = 1
+
+        @event
+        async def set_b(self):
+            await resolving.wait()
+            self.b = 2
+            # Stay out of the flush until set_a has emitted and cleaned.
+            for _ in range(10):
+                await asyncio.sleep(0)
+
+        @event
+        async def both(self):
+            await asyncio.gather(GatheredState.set_a(), GatheredState.set_b())
+
+    async with real_base_state_processor as processor:
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(GatheredState.both())[0]
+            )
+        )
+    sent_b = [
+        delta
+        for _, delta in emitted_deltas
+        if delta.get(GatheredState.get_full_name(), {}).get("b" + FIELD_MARKER) == 2
+    ]
+    assert sent_b, "the gathered handler's change was cleaned without being sent"
+
+
+async def test_gathered_inline_handler_rewrite_of_a_dirty_var_is_kept(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list,
+    token: str,
+):
+    """A gathered handler rewriting a var the other one already dirtied is not lost.
+
+    The dirty-name set does not change in this case, so only a mutation count
+    tells the first handler to pick the new value up before cleaning.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: Deltas emitted to the frontend, in order.
+        token: The client token.
+    """
+    resolving = asyncio.Event()
+
+    class RewriteState(State):
+        a: int = 0
+
+        @rx.var(cache=False)
+        async def slow(self) -> int:
+            if self.a == 1:
+                resolving.set()
+            await asyncio.sleep(0)
+            return self.a
+
+        @event
+        async def set_a(self):
+            self.a = 1
+
+        @event
+        async def rewrite_a(self):
+            await resolving.wait()
+            self.a = 7
+            for _ in range(10):
+                await asyncio.sleep(0)
+
+        @event
+        async def both(self):
+            await asyncio.gather(RewriteState.set_a(), RewriteState.rewrite_a())
+
+    async with real_base_state_processor as processor:
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(RewriteState.both())[0]
+            )
+        )
+    sent_a = [
+        delta[RewriteState.get_full_name()]["a" + FIELD_MARKER]
+        for _, delta in emitted_deltas
+        if "a" + FIELD_MARKER in delta.get(RewriteState.get_full_name(), {})
+    ]
+    assert sent_a[-1] == 7, f"the frontend was left at {sent_a}"
+
+
+async def test_gathered_inline_handler_in_place_change_is_kept(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list,
+    token: str,
+):
+    """A gathered handler's in-place change to a dirty list reaches computed values.
+
+    The list keeps its identity, so only a write count tells the flushing
+    handler to recompute what depends on it before cleaning.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        emitted_deltas: Deltas emitted to the frontend, in order.
+        token: The client token.
+    """
+    resolving = asyncio.Event()
+
+    class InPlaceState(State):
+        items: list[int] = []
+
+        @rx.var
+        def count(self) -> int:
+            return len(self.items)
+
+        @rx.var(cache=False)
+        async def slow(self) -> int:
+            if self.items:
+                resolving.set()
+            await asyncio.sleep(0)
+            return len(self.items)
+
+        @event
+        async def start(self):
+            self.items = [1]
+
+        @event
+        async def append(self):
+            await resolving.wait()
+            self.items.append(2)
+            for _ in range(10):
+                await asyncio.sleep(0)
+
+        @event
+        async def both(self):
+            await asyncio.gather(InPlaceState.start(), InPlaceState.append())
+
+    async with real_base_state_processor as processor:
+        await (
+            await processor.enqueue(
+                token, Event.from_event_type(InPlaceState.both())[0]
+            )
+        )
+    sent_counts = [
+        delta[InPlaceState.get_full_name()]["count" + FIELD_MARKER]
+        for _, delta in emitted_deltas
+        if "count" + FIELD_MARKER in delta.get(InPlaceState.get_full_name(), {})
+    ]
+    assert sent_counts[-1] == 2, f"the frontend was left with counts {sent_counts}"
+
+
+async def test_detached_inline_task_holds_the_lock_until_it_finishes(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A task started from a handler keeps the lock held after the handler returns.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    release = asyncio.Event()
+    seen: dict[str, Any] = {}
+
+    class DetachedState(State):
+        log: list[str] = []
+
+        @event
+        async def child(self):
+            await release.wait()
+            self.log = [*self.log, "child"]
+
+        @event
+        async def parent(self):
+            seen["ctx"] = EventContext.get()
+            asyncio.ensure_future(DetachedState.child())  # noqa: RUF006
+            # Let the child start under the lock before returning.
+            await asyncio.sleep(0)
+            self.log = [*self.log, "parent"]
+
+        @event
+        def next_event(self):
+            self.log = [*self.log, "next"]
+
+    async with real_base_state_processor as processor:
+        parent_future = await processor.enqueue(
+            token, Event.from_event_type(DetachedState.parent())[0]
+        )
+        next_future = await processor.enqueue(
+            token, Event.from_event_type(DetachedState.next_event())[0]
+        )
+        await asyncio.sleep(0.05)
+        # The parent's handler has returned, but its lock is still held for
+        # the detached child, so the next event has not run yet.
+        assert not parent_future.done()
+        assert not next_future.done()
+        assert seen["ctx"].locked_root is not None
+        release.set()
+        await asyncio.wait_for(next_future, 5)
+        assert parent_future.done()
+
+    assert seen["ctx"].locked_root is None
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(DetachedState)).log == ["parent", "child", "next"]
+
+
+async def test_detached_task_started_after_release_is_enqueued(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A task that only starts once the lock is gone runs through the queue instead.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+
+    class LateState(State):
+        log: list[str] = []
+
+        @event
+        def child(self):
+            self.log = [*self.log, "child"]
+
+        @event
+        def parent(self):
+            # Returns before the task gets to run.
+            asyncio.ensure_future(LateState.child())  # noqa: RUF006
+            self.log = [*self.log, "parent"]
+
+        @event
+        def next_event(self):
+            self.log = [*self.log, "next"]
+
+    async with real_base_state_processor as processor:
+        await processor.enqueue(token, Event.from_event_type(LateState.parent())[0])
+        await processor.enqueue(token, Event.from_event_type(LateState.next_event())[0])
+        root_ctx = real_base_state_processor._root_context
+        assert root_ctx is not None
+        log: list[str] = []
+        for _ in range(100):
+            await asyncio.sleep(0.01)
+            state = await root_ctx.state_manager.get_state(
+                BaseStateToken(ident=token, cls=State)
+            )
+            log = (await state.get_state(LateState)).log
+            if len(log) == 3:
+                break
+
+    assert log == ["parent", "next", "child"]
+
+
+async def test_cancelled_background_handler_releases_the_lock_while_waiting_for_inline_runs(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """Cancelling a handler while its lock release waits for a detached run still frees the lock.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    started = asyncio.Event()
+    seen: dict[str, Any] = {}
+
+    class CancelState(State):
+        log: list[str] = []
+
+        @event
+        async def child(self):
+            started.set()
+            await asyncio.Event().wait()
+
+        @event(background=True)
+        async def background(self):
+            async with self:
+                seen["child_task"] = asyncio.ensure_future(CancelState.child())
+                await started.wait()
+            # Leaving `async with self` waits for the detached child.
+
+        @event
+        def ping(self):
+            self.log = [*self.log, "ping"]
+
+    async with real_base_state_processor as processor:
+        bg_future = await processor.enqueue(
+            token, Event.from_event_type(CancelState.background())[0]
+        )
+        await asyncio.wait_for(started.wait(), 5)
+        await asyncio.sleep(0.05)
+        bg_future.cancel()
+        ping_future = await processor.enqueue(
+            token, Event.from_event_type(CancelState.ping())[0]
+        )
+        # The state lock was released despite the cancelled wait.
+        await asyncio.wait_for(ping_future, 5)
+        with contextlib.suppress(asyncio.CancelledError):
+            await asyncio.wait_for(seen["child_task"], 5)
+        assert seen["child_task"].cancelled()
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(CancelState)).log == ["ping"]
+
+
+async def test_release_lock_waits_for_nested_and_late_inline_runs(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+    caplog: pytest.LogCaptureFixture,
+):
+    """The lock outlives a detached run that nests another, spawns a late one, or fails.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+        caplog: The log capture fixture.
+    """
+    release = asyncio.Event()
+
+    class NestedState(State):
+        log: list[str] = []
+
+        @event
+        def inner(self):
+            self.log = [*self.log, "inner"]
+
+        @event
+        async def late(self):
+            await release.wait()
+            self.log = [*self.log, "late"]
+
+        @event
+        async def failing(self):
+            # Still running when the release drains, so it is gathered there.
+            await release.wait()
+            msg = "inline boom"
+            raise ValueError(msg)
+
+        @event
+        async def outer(self):
+            # Nesting an inline run must not drop this task's registration.
+            await NestedState.inner()
+            await asyncio.sleep(0)
+            # Started while the release is already waiting for this run.
+            asyncio.ensure_future(NestedState.late())  # noqa: RUF006
+            asyncio.ensure_future(NestedState.failing())  # noqa: RUF006
+            await asyncio.sleep(0)
+            self.log = [*self.log, "outer"]
+
+        @event
+        async def parent(self):
+            asyncio.ensure_future(NestedState.outer())  # noqa: RUF006
+            await asyncio.sleep(0)
+            self.log = [*self.log, "parent"]
+
+        @event
+        def next_event(self):
+            self.log = [*self.log, "next"]
+
+    with caplog.at_level(logging.ERROR, logger="reflex_base.event.context"):
+        async with real_base_state_processor as processor:
+            await processor.enqueue(
+                token, Event.from_event_type(NestedState.parent())[0]
+            )
+            next_future = await processor.enqueue(
+                token, Event.from_event_type(NestedState.next_event())[0]
+            )
+            await asyncio.sleep(0.05)
+            assert not next_future.done()
+            release.set()
+            await asyncio.wait_for(next_future, 5)
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    # The spawned run starts at the parent's first suspension point and runs
+    # `inner` inline before the parent resumes.
+    assert (await state.get_state(NestedState)).log == [
+        "inner",
+        "parent",
+        "outer",
+        "late",
+        "next",
+    ]
+    assert "inline boom" in caplog.text
+
+
+async def test_handler_cancelled_inside_the_locked_block_cancels_its_inline_runs(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A cancellation delivered inside ``async with self`` still frees the lock.
+
+    The cancelled task is not interrupted again while releasing the lock, so
+    the detached run it would wait for is cancelled instead.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    started = asyncio.Event()
+    seen: dict[str, Any] = {}
+
+    class CancelInsideState(State):
+        log: list[str] = []
+
+        @event
+        async def child(self):
+            started.set()
+            await asyncio.Event().wait()
+
+        @event(background=True)
+        async def background(self):
+            async with self:
+                seen["child_task"] = asyncio.ensure_future(CancelInsideState.child())
+                # Cancelled while still inside the locked block.
+                await asyncio.Event().wait()
+
+        @event
+        def ping(self):
+            self.log = [*self.log, "ping"]
+
+    async with real_base_state_processor as processor:
+        bg_future = await processor.enqueue(
+            token, Event.from_event_type(CancelInsideState.background())[0]
+        )
+        await asyncio.wait_for(started.wait(), 5)
+        bg_future.cancel()
+        ping_future = await processor.enqueue(
+            token, Event.from_event_type(CancelInsideState.ping())[0]
+        )
+        await asyncio.wait_for(ping_future, 5)
+        with contextlib.suppress(asyncio.CancelledError):
+            await asyncio.wait_for(seen["child_task"], 5)
+        assert seen["child_task"].cancelled()
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    state = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    assert (await state.get_state(CancelInsideState)).log == ["ping"]
diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py
index b009404fe8d..3f846eca464 100644
--- a/tests/units/reflex_base/event/processor/test_event_processor.py
+++ b/tests/units/reflex_base/event/processor/test_event_processor.py
@@ -26,7 +26,7 @@
 from reflex_base.event.processor.future import EventFuture
 from reflex_base.registry import RegistrationContext
 
-from reflex.event import Event, EventHandler
+from reflex.event import Event, EventHandler, EventSpec
 from tests.units.conftest import active_tracer
 
 # Module-level log so event handlers can record what happened.
@@ -1901,3 +1901,59 @@ async def test_top_level_event_under_local_span_names_no_parent_event(
         == top.attributes[otel.ATTR_EVENT_TXID]
         == future.txid
     )
+
+
+async def test_bare_handler_return_value_resolves_future(
+    mock_event_processor: EventProcessor,
+    token: str,
+):
+    """A bare handler's return value resolves its future.
+
+    A returned EventSpec is a description of an event, not work for the base
+    processor to run, so the handler it names must not be executed.
+
+    Args:
+        mock_event_processor: The event processor with mock root context.
+        token: The client token.
+    """
+
+    def _returns_value() -> int:
+        return 7
+
+    async def _returns_spec() -> EventSpec:  # noqa: RUF029
+        return logging_event("not run")
+
+    value_event = EventHandler(fn=_returns_value)
+    spec_event = EventHandler(fn=_returns_spec)
+    RegistrationContext.register_event_handler(value_event)
+    RegistrationContext.register_event_handler(spec_event)
+    async with mock_event_processor as ep:
+        value_future = await ep.enqueue(token, Event.from_event_type(value_event())[0])
+        assert await value_future == 7
+        spec_future = await ep.enqueue(token, Event.from_event_type(spec_event())[0])
+        spec = await spec_future
+        assert isinstance(spec, EventSpec)
+    assert _CALL_LOG == []
+
+
+async def test_bare_handler_returned_future_is_awaited(
+    mock_event_processor: EventProcessor,
+    token: str,
+):
+    """A bare handler returning a Future still has it awaited; only event specs are not.
+
+    Args:
+        mock_event_processor: The event processor with mock root context.
+        token: The client token.
+    """
+
+    def _returns_future() -> asyncio.Future[int]:
+        future: asyncio.Future[int] = asyncio.get_running_loop().create_future()
+        asyncio.get_running_loop().call_soon(future.set_result, 11)
+        return future
+
+    future_event = EventHandler(fn=_returns_future)
+    RegistrationContext.register_event_handler(future_event)
+    async with mock_event_processor as ep:
+        event_future = await ep.enqueue(token, Event.from_event_type(future_event())[0])
+        assert await event_future == 11
diff --git a/tests/units/reflex_base/event/test_context.py b/tests/units/reflex_base/event/test_context.py
index 46558b0bf04..26656119f54 100644
--- a/tests/units/reflex_base/event/test_context.py
+++ b/tests/units/reflex_base/event/test_context.py
@@ -98,3 +98,21 @@ async def test_emit_event_noop_when_no_impl():
         emit_event_impl=None,
     )
     await ctx.emit_event()
+
+
+async def test_enqueue_returns_futures(mock_root_event_context: EventContext):
+    """Enqueue returns whatever enqueue_impl returns, so callers can await the futures.
+
+    Args:
+        mock_root_event_context: The root event context fixture.
+    """
+    futures = [mock.sentinel.future]
+
+    async def enqueue_impl(token: str, *events: object) -> list:  # noqa: RUF029
+        assert token == "tok"
+        return futures
+
+    ctx = dataclasses.replace(
+        mock_root_event_context.fork(token="tok"), enqueue_impl=enqueue_impl
+    )
+    assert await ctx.enqueue(mock.sentinel.event) is futures
diff --git a/tests/units/reflex_base/vars/test_base.py b/tests/units/reflex_base/vars/test_base.py
index 242cf2e85a9..c9045d71638 100644
--- a/tests/units/reflex_base/vars/test_base.py
+++ b/tests/units/reflex_base/vars/test_base.py
@@ -564,6 +564,33 @@ def create(cls, value: Coordinate, _var_data=None):
         base._literal_var_by_type.clear()
 
 
+def test_literal_var_dispatch_sees_through_proxies(clean_registration_context):
+    """Proxies of different values dispatch on the wrapped value, not the proxy type.
+
+    Args:
+        clean_registration_context: An isolated state registry.
+    """
+    from reflex_base.vars import base
+
+    state_cls = type(
+        "ProxyDispatchState",
+        (BaseState,),
+        {
+            "__module__": __name__,
+            "__annotations__": {"mapping": dict[str, int], "items": list[int]},
+            "mapping": {"a": 1},
+            "items": [1, 2],
+        },
+    )
+    state = state_cls()
+    try:
+        assert str(LiteralVar.create(state.mapping)) == '({ ["a"] : 1 })'  # pyright: ignore [reportAttributeAccessIssue]
+        assert str(LiteralVar.create(state.items)) == "[1, 2]"  # pyright: ignore [reportAttributeAccessIssue]
+        assert MutableProxy not in base._literal_var_by_type
+    finally:
+        base._literal_var_by_type.clear()
+
+
 def test_guess_type_dispatch_follows_later_registrations():
     """A Var subclass registered after a guess wins the next guess for its type."""
 
diff --git a/tests/units/test_app.py b/tests/units/test_app.py
index 93d136f8885..0a6b626e506 100644
--- a/tests/units/test_app.py
+++ b/tests/units/test_app.py
@@ -6031,6 +6031,34 @@ def reader():
     assert json.loads(marker.read_text()) == routes
 
 
+async def test_await_inside_app_modify_state_runs_inline(app_with_processor: App):
+    """A handler awaited inside ``modify_state`` runs on the locked state.
+
+    Args:
+        app_with_processor: An App with a mock event processor and root context.
+    """
+
+    class ModifyAwaitState(BaseState):
+        n: int = 0
+
+        @rx.event
+        def bump(self, by: int) -> int:
+            self.n += by
+            return self.n
+
+    app_with_processor._state_manager = StateManagerMemory()
+    app_with_processor._event_namespace = AsyncMock()
+    async with app_with_processor.modify_state(
+        BaseStateToken(ident="token-C", cls=ModifyAwaitState)
+    ) as state:
+        assert EventContext.get().locked_root is state._get_root_state()
+        assert await ModifyAwaitState.bump(2) == 2
+        assert await ModifyAwaitState.bump(3) == 5
+        assert (await state.get_state(ModifyAwaitState)).n == 5
+    with pytest.raises(LookupError):
+        EventContext.get()
+
+
 @pytest.mark.parametrize(
     ("prod", "config_kwargs", "expected"),
     [
diff --git a/tests/units/test_event.py b/tests/units/test_event.py
index abc07d9a963..49a32697f03 100644
--- a/tests/units/test_event.py
+++ b/tests/units/test_event.py
@@ -1713,3 +1713,95 @@ def handler():
     assert event_sequence(events_user_list) is events_user_list
     assert event_sequence("abc") == ["abc"]
     assert event_sequence(b"ab") == [b"ab"]
+
+
+def test_decentralized_event_actions_are_registered():
+    """The registered handler of a decentralized event carries its event actions."""
+
+    class S(BaseState):
+        field: Field[str] = field("")
+
+    @event(stop_propagation=True)
+    def e(s: S):
+        s.field = "foo"
+
+    assert isinstance(e, EventHandler)
+    registered = RegistrationContext.get().event_handlers[
+        format.format_event_handler(e)
+    ]
+    assert registered.handler is e
+    assert registered.handler.event_actions == {"stopPropagation": True}
+    assert registered.states == (S,)
+
+
+def test_rx_event_stateless_registers_globally():
+    """A function with no state parameter registers as a stateless handler."""
+
+    @event
+    def ping(x: str) -> str:
+        return x
+
+    assert isinstance(ping, EventHandler)
+    assert ping.state is None
+    registered = RegistrationContext.get().event_handlers[
+        format.format_event_handler(ping)
+    ]
+    assert registered.handler is ping
+    assert registered.states == ()
+    spec = ping("1")
+    assert isinstance(spec, EventSpec)
+    [ev] = Event.from_event_type(spec)
+    assert ev.name == format.format_event_handler(ping)
+    assert ev.payload == {"x": "1"}
+    _ = rx.input(on_change=ping)
+
+
+def test_rx_event_stateless_with_generic_annotation():
+    """A stateless handler may annotate its first parameter with a generic alias."""
+
+    @event
+    def summarize(items: list[int], label: str | None = None) -> int:
+        return sum(items)
+
+    assert isinstance(summarize, EventHandler)
+    assert summarize.state is None
+    registered = RegistrationContext.get().event_handlers[
+        format.format_event_handler(summarize)
+    ]
+    assert registered.states == ()
+
+
+def test_event_handler_and_spec_expose_wrapped_function() -> None:
+    """EventHandler and EventSpec expose the decorated function as __wrapped__."""
+
+    class WrappedState(BaseState):
+        @rx.event
+        def handle(self, value: int):
+            pass
+
+    handler = WrappedState.handle
+    assert isinstance(handler, EventHandler)
+    assert handler.__wrapped__ is handler.fn
+    assert handler.__wrapped__.__name__ == "handle"
+    assert handler(1).__wrapped__ is handler.fn
+
+
+def test_rx_event_on_a_non_state_class_method_still_calls_it() -> None:
+    """@rx.event on a method of a class that is not a state warns and stays a method."""
+    from unittest.mock import patch
+
+    from reflex_base.utils import console
+
+    with patch.object(console, "deprecate", wraps=console.deprecate) as mock_deprecate:
+
+        class Helper:
+            @rx.event
+            def compute(self, x: int) -> int:
+                return x * 2
+
+    assert (
+        mock_deprecate.call_args.kwargs["feature_name"]
+        == "@rx.event on a method of a class that is not a state"
+    )
+    assert isinstance(Helper.compute, EventHandler)
+    assert Helper().compute(3) == 6  # pyright: ignore [reportArgumentType]
diff --git a/tests/units/test_state.py b/tests/units/test_state.py
index 0334c07ad48..40d544df772 100644
--- a/tests/units/test_state.py
+++ b/tests/units/test_state.py
@@ -7500,6 +7500,220 @@ def _clean(self):
     assert not storage.dirty_vars
 
 
+def test_rx_event_handler_bound_at_class_creation() -> None:
+    """A decorated method is an EventHandler bound to its class when the body runs."""
+
+    class BoundState(BaseState):
+        @rx.event
+        def handle(self, value: int):
+            pass
+
+    handler = BoundState.__dict__["handle"]
+    assert isinstance(handler, EventHandler)
+    assert handler.state is BoundState
+    assert handler._type_hints == {"value": int}
+    assert BoundState.event_handlers["handle"] is handler
+    registered = RegistrationContext.get().event_handlers[
+        f"{BoundState.get_full_name()}.handle"
+    ]
+    assert registered.handler is handler
+    assert registered.states == (BoundState,)
+
+
+def test_mixin_decorated_handler_copied_per_state() -> None:
+    """A mixin's decorated handler is bound to the mixin; each state gets its own copy."""
+
+    class CopyMixin(BaseState, mixin=True):
+        @rx.event(background=True, stop_propagation=True)
+        async def work(self):
+            pass
+
+    class UsesCopyMixinA(CopyMixin, State):
+        pass
+
+    class UsesCopyMixinB(CopyMixin, State):
+        pass
+
+    template = CopyMixin.__dict__["work"]
+    assert isinstance(template, EventHandler)
+    assert template.state is CopyMixin
+    handler_a = UsesCopyMixinA.event_handlers["work"]
+    handler_b = UsesCopyMixinB.event_handlers["work"]
+    assert handler_a.state is UsesCopyMixinA
+    assert handler_b.state is UsesCopyMixinB
+    assert len({template.fn, handler_a.fn, handler_b.fn}) == 3
+    for handler in (handler_a, handler_b):
+        assert handler.is_background
+        assert handler.event_actions == {"stopPropagation": True}
+
+
+def test_aliased_handler_keeps_its_state() -> None:
+    """Assigning another state's handler in a class body does not rebind it."""
+
+    class OwnerState(BaseState):
+        @rx.event
+        def handle(self):
+            pass
+
+    class AliasState(BaseState):
+        alias = OwnerState.handle
+
+    owner_handler = OwnerState.__dict__["handle"]
+    assert isinstance(owner_handler, EventHandler)
+    assert owner_handler.state is OwnerState
+    assert AliasState.__dict__["alias"] is owner_handler
+    assert "alias" not in AliasState.event_handlers
+
+
+def _logged(fn):
+    """Wrap a method the way an auth decorator stacked above @rx.event does.
+
+    Args:
+        fn: The decorated method (an EventHandler when stacked above @rx.event).
+
+    Returns:
+        The wrapper, which calls ``fn`` with the instance like a plain method.
+    """
+
+    @functools.wraps(fn)
+    def wrapper(self, *args, **kwargs):
+        return fn(self, *args, **kwargs)
+
+    return wrapper
+
+
+def test_wrapper_above_rx_event_is_deprecated_and_still_runs() -> None:
+    """A decorator above @rx.event warns and keeps working as the plain method it wraps."""
+    from reflex_base.utils import console as base_console
+
+    from reflex.utils import console
+
+    with (
+        patch.object(console, "deprecate", wraps=console.deprecate) as state_deprecate,
+        patch.object(
+            base_console, "deprecate", wraps=base_console.deprecate
+        ) as event_deprecate,
+    ):
+
+        class WrappedState(BaseState):
+            x: int = 0
+
+            @_logged
+            @rx.event
+            def inc(self, by: int):
+                self.x += by
+
+            @_logged
+            @rx.event(background=True)
+            async def bg(self):
+                pass
+
+        assert (
+            state_deprecate.call_args.kwargs["feature_name"]
+            == "Decorators above @rx.event"
+        )
+        handler = WrappedState.inc
+        assert isinstance(handler, EventHandler)
+        assert handler.state is WrappedState
+        assert format.format_event_handler(handler).endswith(".inc")
+        assert WrappedState.event_handlers["inc"] is handler
+        # The payload coercion still sees the method's annotations.
+        assert handler._get_type_hints()["by"] is int
+        assert handler.fn.__module__ == __name__
+        assert WrappedState.bg.is_background  # pyright: ignore [reportAttributeAccessIssue]
+        state = WrappedState()
+        # The wrapper calls the inner handler with the instance, like on main.
+        handler.fn(state, 2)
+        assert state.x == 2
+        assert (
+            event_deprecate.call_args.kwargs["feature_name"]
+            == "Calling an event handler with a state instance"
+        )
+
+
+def test_stacked_wrappers_above_rx_event_still_run() -> None:
+    """Decorators stacked above @rx.event are found through the whole __wrapped__ chain."""
+    from reflex.utils import console
+
+    with patch.object(console, "deprecate", wraps=console.deprecate) as mock_deprecate:
+
+        class StackedState(BaseState):
+            x: int = 0
+
+            @_logged
+            @_logged
+            @rx.event
+            def inc(self, by: int):
+                self.x += by
+
+        assert (
+            mock_deprecate.call_args.kwargs["feature_name"]
+            == "Decorators above @rx.event"
+        )
+        handler = StackedState.inc
+        assert isinstance(handler, EventHandler)
+        assert format.format_event_handler(handler).endswith(".inc")
+        state = StackedState()
+        handler.fn(state, 3)
+        assert state.x == 3
+
+
+def test_decorated_handler_shadowing_builtin_raises() -> None:
+    """A decorated handler that shadows a State method is rejected like a plain one."""
+    with pytest.raises(NameError, match="shadows a builtin State method"):
+
+        class ShadowState(BaseState):
+            @rx.event
+            def reset(self):
+                pass
+
+
+def test_rx_event_on_static_and_class_methods_is_deprecated_and_ignored() -> None:
+    """@rx.event on a staticmethod or classmethod warns and leaves it as it is."""
+    from reflex_base.utils import console
+
+    with patch.object(console, "deprecate", wraps=console.deprecate) as mock_deprecate:
+
+        class StaticState(BaseState):
+            @rx.event
+            @staticmethod
+            def helper(value: int) -> int:
+                return value * 2
+
+    assert (
+        mock_deprecate.call_args.kwargs["feature_name"]
+        == "@rx.event on a staticmethod or classmethod"
+    )
+    assert isinstance(StaticState.__dict__["helper"], staticmethod)
+    assert StaticState.helper(2) == 4
+
+
+def test_decorated_handler_on_state_derived_from_a_mixin_only() -> None:
+    """A state whose base flips the mixin flag on (like ``rx.SharedState``) owns its handlers.
+
+    The metaclass first marks a class declared on mixins alone as a mixin and
+    the base's ``__init_subclass__`` flips it, so binding must not depend on
+    that flag.
+    """
+
+    class FlippingMixin(BaseState, mixin=True):
+        def __init_subclass__(cls, **kwargs):
+            kwargs["mixin"] = False
+            cls._mixin = False
+            super().__init_subclass__(**kwargs)
+
+    class FromFlippingMixin(FlippingMixin):
+        @rx.event
+        def handle(self):
+            pass
+
+    assert not FromFlippingMixin._mixin
+    handler = FromFlippingMixin.__dict__["handle"]
+    assert isinstance(handler, EventHandler)
+    assert handler.state is FromFlippingMixin
+    assert FromFlippingMixin.event_handlers["handle"] is handler
+
+
 async def test_shared_storage_name_syncs_siblings(
     clean_registration_context: RegistrationContext,
 ):