From bda6fb5a3d4a80ba5bfe0739b91324065a7cbfbe Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:25:13 +0000 Subject: [PATCH 01/13] Resolve event futures with handler return values and bind @rx.event handlers at class creation A handler's non-event return value now resolves its EventFuture instead of raising TypeError; returned events still chain. @rx.event returns the EventHandler itself, bound to its state class by __set_name__, and a function without a state parameter registers as a stateless handler that runs without the state lock and outside the per-token FIFO. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/__init__.py | 157 +++++++--- .../src/reflex_base/event/context.py | 15 +- .../event/processor/base_state_processor.py | 114 +++++--- .../event/processor/event_processor.py | 36 ++- .../reflex-base/src/reflex_base/vars/base.py | 5 + reflex/state.py | 24 +- .../processor/test_base_state_processor.py | 267 ++++++++++++++++++ .../event/processor/test_event_processor.py | 35 ++- tests/units/reflex_base/event/test_context.py | 18 ++ tests/units/test_event.py | 41 +++ tests/units/test_state.py | 105 +++++++ 11 files changed, 719 insertions(+), 98 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/__init__.py b/packages/reflex-base/src/reflex_base/event/__init__.py index 7f6c6ffe43e..5e1d7568d24 100644 --- a/packages/reflex-base/src/reflex_base/event/__init__.py +++ b/packages/reflex-base/src/reflex_base/event/__init__.py @@ -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 @@ -655,6 +640,27 @@ def supersedes(self) -> bool: """ return getattr(self.fn, SUPERSEDES_MARKER, False) + 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) and one declared on a mixin + (copied per concrete state instead) are left unbound. + + Args: + owner: The class the handler is declared on. + name: The attribute name. + """ + del name + if self.state is not None or getattr(owner, "_mixin", True): + return + from reflex.state import BaseState + + if isinstance(owner, type) and issubclass(owner, BaseState): + object.__setattr__(self, "state", owner) + self._get_type_hints() + def __get__(self, instance: Any, owner: type | None = None) -> Any: """Get the handler on class access, or its function bound to a state. @@ -831,6 +837,26 @@ def __init__( object.__setattr__(self, "client_handler_name", client_handler_name) object.__setattr__(self, "args", args or ()) + 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. @@ -3119,6 +3145,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 +3240,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 +3256,11 @@ 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[[BASE_STATE, Unpack[P]], Any] | None" = None, + func: "Callable[[Unpack[P]], Any]", *, background: bool | None = None, supersedes: bool | None = None, @@ -3221,7 +3269,20 @@ 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[*P]: ... + + def __new__( + cls, + func: "Callable[..., Any] | None" = None, + *, + 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 | EventDecorator": """Wrap a function to be used as an event. Args: @@ -3272,9 +3333,10 @@ 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)): + msg = "Event handlers cannot be static or class methods." + raise TypeError(msg) if background is True: if not inspect.iscoroutinefunction( func @@ -3288,48 +3350,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] == "" ): + # 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) + name = ( + (func.__module__ + "." + qualname) + .replace(".", "_") + .replace("", "_") + .removeprefix("_") + ) + object.__setattr__(func, "__name__", name) + object.__setattr__(func, "__qualname__", name) if state_cls and issubclass(state_cls, BaseState): - name = ( - (func.__module__ + "." + qualname) - .replace(".", "_") - .replace("", "_") - .removeprefix("_") - ) - object.__setattr__(func, "__name__", name) - object.__setattr__(func, "__qualname__", name) 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..8b77e90228d 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -5,7 +5,7 @@ import dataclasses import functools 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 @@ -17,6 +17,7 @@ from reflex.istate.manager import StateManager from reflex_base.event import Event + from reflex_base.event.processor.future import EventFuture @functools.lru_cache @@ -43,12 +44,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. """ ... @@ -163,10 +167,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..bddda41f744 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 @@ -66,6 +66,28 @@ 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 _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 +101,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: @@ -282,7 +294,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. @@ -311,15 +323,15 @@ def ensure_locked( 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 +339,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 +350,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 +367,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 +381,23 @@ 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 is the result. + if events is None or _is_event_like(events): await chain_updates( events, root_state=ensure_locked(state, root_state), handler_name=handler_name, ) + return None + if inspect.iscoroutine(events): + # A coroutine that was never awaited is a bug, not a result. + _check_valid_yield(events, handler_name=handler_name) + await chain_updates( + None, root_state=ensure_locked(state, root_state), handler_name=handler_name + ) + return events class BaseStateEventProcessor(EventProcessor): @@ -528,9 +544,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 +569,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,7 +627,7 @@ 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]) @@ -597,13 +638,12 @@ async def _execute_event( # Process non-background events while holding the lock. if not registered_handler.handler.is_background: - await process_event( + return await process_event( handler=registered_handler.handler, payload=event.payload, state=substate, root_state=root_state, ) - return # 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 +658,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 +700,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 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 51f7e1231be..8f04458c097 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 @@ -701,9 +701,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 @@ -718,15 +729,21 @@ 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 + # Only a coroutine is run for the handler: an awaitable EventSpec it + # returns is a description of an event, not work to do here. + if inspect.iscoroutine(result): + 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 @@ -741,16 +758,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 ) @@ -866,8 +885,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..c010903b219 100644 --- a/packages/reflex-base/src/reflex_base/vars/base.py +++ b/packages/reflex-base/src/reflex_base/vars/base.py @@ -4701,7 +4701,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/state.py b/reflex/state.py index 9859e0d3c2a..5e1f2c152f2 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -803,7 +803,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 +869,12 @@ def _bind_mixin_members(cls) -> None: _var_data=VarData.from_state(cls), ), ) + elif isinstance(value, EventHandler) and value.state is None: + # A decorated handler of a 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 +929,18 @@ def _item_is_event_handler(name: str, value: Any) -> bool: Returns: Whether the item is an event handler. + + Raises: + TypeError: If a decorator was applied on top of ``@rx.event``. """ + if isinstance(getattr(value, "__wrapped__", None), EventHandler): + # Wrapped as a handler, the wrapper would return the event spec + # of the handler it wraps, chaining that handler forever. + msg = ( + f"Event handler `{name}` is decorated above @rx.event; apply " + "other decorators below it." + ) + raise TypeError(msg) return ( not name.startswith("_") and isinstance(value, Callable) 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 ebbbc91f23d..31c22d38140 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 @@ -3430,3 +3430,270 @@ 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_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"] 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 e7a389b4392..1eae8ffe638 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -22,7 +22,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. @@ -1784,3 +1784,36 @@ 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 == [] 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/test_event.py b/tests/units/test_event.py index abc07d9a963..a3c6d7ee7a9 100644 --- a/tests/units/test_event.py +++ b/tests/units/test_event.py @@ -1713,3 +1713,44 @@ 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) diff --git a/tests/units/test_state.py b/tests/units/test_state.py index cc61b0de658..b8fe334d2a7 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -7539,3 +7539,108 @@ def _clean(self): assert storage.value == "fresh" assert storage._previous_dirty_vars >= {"snapshot_write", "value"} 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 stays unbound; 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 None + 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 test_wrapper_above_rx_event_raises() -> None: + """A decorator applied on top of @rx.event is rejected at class creation.""" + + def logged(fn): + @functools.wraps(fn) + def wrapper(*args, **kwargs): + return fn(*args, **kwargs) + + return wrapper + + with pytest.raises(TypeError, match=r"decorated above @rx\.event"): + + class WrappedState(BaseState): + @logged + @rx.event + def handle(self): + pass + + +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_rejects_static_and_class_methods() -> None: + """@rx.event on a staticmethod or classmethod raises.""" + with pytest.raises(TypeError, match="static or class methods"): + + class StaticState(BaseState): + @rx.event + @staticmethod + def handle(): + pass From 40bb85f9e95c3a9ecb99f554fbbeee566dd66723 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:25:14 +0000 Subject: [PATCH 02/13] Make EventHandler and EventSpec awaitable Awaiting a handler (or a spec with bound args) runs it in the active EventContext: inline on the locked state tree while the current task holds the token's state lock, which the context now records, or enqueued and awaited when no lock is held. Its yields emit deltas and chain as usual, and the handler's return value is the result of the await. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/__init__.py | 54 ++- .../src/reflex_base/event/context.py | 37 +- .../event/processor/base_state_processor.py | 97 +++- reflex/app.py | 9 +- reflex/istate/proxy.py | 3 + tests/units/istate/test_proxy.py | 18 + .../processor/test_base_state_processor.py | 435 ++++++++++++++++++ tests/units/test_app.py | 28 ++ 8 files changed, 659 insertions(+), 22 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/__init__.py b/packages/reflex-base/src/reflex_base/event/__init__.py index 5e1d7568d24..70ec79280a8 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, @@ -640,6 +640,14 @@ def supersedes(self) -> bool: """ return getattr(self.fn, SUPERSEDES_MARKER, False) + 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. @@ -837,6 +845,34 @@ 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) + + 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. @@ -2908,12 +2944,20 @@ 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] + 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. diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index 8b77e90228d..d1980488167 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -2,10 +2,11 @@ from __future__ import annotations +import contextlib import dataclasses import functools import os -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from typing import TYPE_CHECKING, Any, Protocol from reflex_base import otel @@ -16,6 +17,7 @@ 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 @@ -113,8 +115,13 @@ 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 ) # OpenTelemetry context active when this event was enqueued (None when tracing is off). otel_context: Context | None = dataclasses.field(default=None, repr=False) @@ -147,6 +154,30 @@ 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) + + @contextlib.contextmanager + def holding_lock(self, root_state: BaseState) -> Iterator[None]: + """Mark the state lock as held for the duration of the block. + + Args: + root_state: The root of the locked state tree. + + Yields: + Nothing. + """ + self.set_locked_root(root_state) + try: + yield + finally: + self.set_locked_root(None) + async def emit_delta(self, delta: Mapping[str, Mapping[str, Any]]) -> None: """Emit a delta to the frontend. 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 bddda41f744..67cbeef5251 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 @@ -320,6 +320,73 @@ 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 + + name = format_event_handler(spec.handler) + try: + ctx = EventContext.get() + except LookupError as le: + msg = ( + f"Event handler `{name}` 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 + if spec.client_handler_name: + msg = f"Upload event `{name}` cannot be awaited; use it as an event trigger." + raise EventHandlerTypeError(msg) + event = spec._to_event(ctx.router_data) + registered = RegistrationContext.get().event_handlers.get(event.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 + if not registered.states: + return await process_event( + handler=handler, payload=event.payload, state=None, root_state=None + ) + root_state = ctx.locked_root + if root_state is None: + (future,) = await ctx.enqueue(event) + return await future + if 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) + substate = await root_state.get_state(registered.states[0]) + return await process_event( + handler=handler, payload=event.payload, state=substate, root_state=root_state + ) + + async def process_event( handler: EventHandler, payload: dict, @@ -633,17 +700,18 @@ async def _execute_event( 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: - return await process_event( - handler=registered_handler.handler, - payload=event.payload, - state=substate, - root_state=root_state, - ) + with ctx.holding_lock(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: + return await process_event( + handler=registered_handler.handler, + payload=event.payload, + state=substate, + root_state=root_state, + ) # 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) @@ -725,4 +793,9 @@ async def _handle_backend_exception( ) -__all__ = ["BaseStateEventProcessor", "chain_updates", "process_event"] +__all__ = [ + "BaseStateEventProcessor", + "chain_updates", + "execute_event_inline", + "process_event", +] diff --git a/reflex/app.py b/reflex/app.py index 97d7d7472e5..9c37e9df036 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -1888,7 +1888,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) @@ -1897,7 +1897,12 @@ 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 not None: + # An awaited handler runs inline on the locked tree. + with forked_context.holding_lock(state._get_root_state()): + yield state + else: + yield state delta = await state._get_resolved_delta() state._clean() if delta: diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py index a4df7c2401c..b023fa43701 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,6 +327,7 @@ async def __aexit__(self, *exc_info: Any) -> None: finally: try: if self._self_mutable and self._self_actx is not None: + EventContext.get().set_locked_root(None) await self._self_actx.__aexit__(*exc_info) finally: self._self_actx = None 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 31c22d38140..ea24a219f3d 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 @@ -3697,3 +3697,438 @@ def 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_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 diff --git a/tests/units/test_app.py b/tests/units/test_app.py index ae575628094..8141e35b555 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -5795,3 +5795,31 @@ def reader(): for future in readers: future.result() 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() From c224855c44d241c5205d29dfe96146faf74e89d4 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:25:14 +0000 Subject: [PATCH 03/13] Document awaitable and stateless event handlers Adds the docs, news fragments and integration scenarios for awaiting event handlers from foreground and background handlers. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- docs/events/background_events.md | 1 + docs/events/chaining_events.md | 52 ++++++++++++++++++- docs/events/decentralized_event_handlers.md | 25 +++++++++ news/+awaitable-event-handlers.feature.md | 13 +++++ .../news/+awaitable-event-handlers.feature.md | 13 +++++ .../src/reflex_base/event/__init__.py | 7 +-- reflex/state.py | 4 +- tests/integration/test_background_task.py | 46 ++++++++++++++++ tests/integration/test_event_chain.py | 28 ++++++++++ tests/units/test_state.py | 30 ++++++++++- 10 files changed, 211 insertions(+), 8 deletions(-) create mode 100644 news/+awaitable-event-handlers.feature.md create mode 100644 packages/reflex-base/news/+awaitable-event-handlers.feature.md diff --git a/docs/events/background_events.md b/docs/events/background_events.md index 73d7b1e27bc..abcaf70f1a8 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 anywhere 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` it runs right away on the locked state. 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..d457a344b54 100644 --- a/docs/events/chaining_events.md +++ b/docs/events/chaining_events.md @@ -47,11 +47,61 @@ 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 events it yields or returns are +queued after the current event, as if the caller had yielded them. + +```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; they run one after the other on the same state. +``` + ## 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..a3799bafa4b 100644 --- a/docs/events/decentralized_event_handlers.md +++ b/docs/events/decentralized_event_handlers.md @@ -159,3 +159,28 @@ 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, can be used +as an event trigger, and can be awaited from other handlers, but it runs +without a state and without taking the state lock, so it never waits behind +the other events of the same client. + +```python box +@rx.event +async def fetch_quote(symbol: str) -> float: + return await price_service.quote(symbol) + + +class PortfolioState(rx.State): + price: float = 0.0 + + @rx.event + async def refresh(self, symbol: str): + self.price = await fetch_quote(symbol) +``` + +A stateless handler can still yield or return events, which run against the +state of the client that triggered it. diff --git a/news/+awaitable-event-handlers.feature.md b/news/+awaitable-event-handlers.feature.md new file mode 100644 index 00000000000..774f7e5e995 --- /dev/null +++ b/news/+awaitable-event-handlers.feature.md @@ -0,0 +1,13 @@ +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, and a handler may now return any value rather than only events. `@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.feature.md b/packages/reflex-base/news/+awaitable-event-handlers.feature.md new file mode 100644 index 00000000000..774f7e5e995 --- /dev/null +++ b/packages/reflex-base/news/+awaitable-event-handlers.feature.md @@ -0,0 +1,13 @@ +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, and a handler may now return any value rather than only events. `@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/src/reflex_base/event/__init__.py b/packages/reflex-base/src/reflex_base/event/__init__.py index 70ec79280a8..40378942b62 100644 --- a/packages/reflex-base/src/reflex_base/event/__init__.py +++ b/packages/reflex-base/src/reflex_base/event/__init__.py @@ -653,15 +653,16 @@ def __set_name__(self, owner: type, name: str) -> None: 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) and one declared on a mixin - (copied per concrete state instead) are left unbound. + 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. """ del name - if self.state is not None or getattr(owner, "_mixin", True): + if self.state is not None: return from reflex.state import BaseState diff --git a/reflex/state.py b/reflex/state.py index 5e1f2c152f2..4936ed2c47c 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -869,8 +869,8 @@ def _bind_mixin_members(cls) -> None: _var_data=VarData.from_state(cls), ), ) - elif isinstance(value, EventHandler) and value.state is None: - # A decorated handler of a mixin: copy its function, which + 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}" 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/test_state.py b/tests/units/test_state.py index b8fe334d2a7..aeddc25cae9 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -7562,7 +7562,7 @@ def handle(self, value: int): def test_mixin_decorated_handler_copied_per_state() -> None: - """A mixin's decorated handler stays unbound; each state gets its own copy.""" + """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) @@ -7577,7 +7577,7 @@ class UsesCopyMixinB(CopyMixin, State): template = CopyMixin.__dict__["work"] assert isinstance(template, EventHandler) - assert template.state is None + assert template.state is CopyMixin handler_a = UsesCopyMixinA.event_handlers["work"] handler_b = UsesCopyMixinB.event_handlers["work"] assert handler_a.state is UsesCopyMixinA @@ -7644,3 +7644,29 @@ class StaticState(BaseState): @staticmethod def handle(): pass + + +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 From 77b086950f0819995ce3fcd4c1932c7507517471 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:25:14 +0000 Subject: [PATCH 04/13] Format news fragments and name the event once in the inline executor Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- news/+awaitable-event-handlers.feature.md | 1 + .../news/+awaitable-event-handlers.feature.md | 1 + .../event/processor/base_state_processor.py | 12 ++++++------ 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/news/+awaitable-event-handlers.feature.md b/news/+awaitable-event-handlers.feature.md index 774f7e5e995..47b4584ea77 100644 --- a/news/+awaitable-event-handlers.feature.md +++ b/news/+awaitable-event-handlers.feature.md @@ -6,6 +6,7 @@ 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) diff --git a/packages/reflex-base/news/+awaitable-event-handlers.feature.md b/packages/reflex-base/news/+awaitable-event-handlers.feature.md index 774f7e5e995..47b4584ea77 100644 --- a/packages/reflex-base/news/+awaitable-event-handlers.feature.md +++ b/packages/reflex-base/news/+awaitable-event-handlers.feature.md @@ -6,6 +6,7 @@ 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) 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 67cbeef5251..88c7a7237ee 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 @@ -345,21 +345,21 @@ async def execute_event_inline(spec: EventSpec) -> Any: from reflex_base.registry import RegistrationContext from reflex_base.utils.exceptions import EventHandlerTypeError - name = format_event_handler(spec.handler) try: ctx = EventContext.get() except LookupError as le: msg = ( - f"Event handler `{name}` can only be awaited from an event handler " - "or inside `app.modify_state`; use it as an event trigger or yield " - "it instead." + 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) - event = spec._to_event(ctx.router_data) - registered = RegistrationContext.get().event_handlers.get(event.name) + registered = RegistrationContext.get().event_handlers.get(name) if registered is None: msg = ( f"`{name}` is a frontend-only event, so it cannot be awaited; " From daeb277e3daf66edf872994cbfeadc7083754a6d Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:35:21 +0000 Subject: [PATCH 05/13] Address review on awaitable event handlers Raise the reflex-base floor to 0.10.2.dev0 for the new EventContext lock methods; keep the state lock held until inline handler runs started from the handler finish; accept generic annotations on stateless handlers; keep awaiting futures returned by bare handlers; tighten the docs and news fragments and add a breaking fragment for the decorator ordering change. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- docs/events/background_events.md | 2 +- docs/events/chaining_events.md | 8 +- news/+awaitable-event-handlers.breaking.md | 1 + news/+awaitable-event-handlers.feature.md | 2 +- .../+awaitable-event-handlers.breaking.md | 1 + .../news/+awaitable-event-handlers.feature.md | 2 +- .../src/reflex_base/event/__init__.py | 2 +- .../src/reflex_base/event/context.py | 32 ++++- .../event/processor/base_state_processor.py | 24 +++- .../event/processor/event_processor.py | 11 +- pyproject.toml | 2 +- reflex/app.py | 2 +- reflex/istate/proxy.py | 2 +- .../processor/test_base_state_processor.py | 109 ++++++++++++++++++ .../event/processor/test_event_processor.py | 23 ++++ tests/units/test_event.py | 15 +++ 16 files changed, 215 insertions(+), 23 deletions(-) create mode 100644 news/+awaitable-event-handlers.breaking.md create mode 100644 packages/reflex-base/news/+awaitable-event-handlers.breaking.md diff --git a/docs/events/background_events.md b/docs/events/background_events.md index abcaf70f1a8..a779a1c9997 100644 --- a/docs/events/background_events.md +++ b/docs/events/background_events.md @@ -196,5 +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 anywhere 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` it runs right away on the locked state. See [Awaiting Event Handlers](/docs/events/chaining-events/#awaiting-event-handlers). +- 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 d457a344b54..5c0649e3428 100644 --- a/docs/events/chaining_events.md +++ b/docs/events/chaining_events.md @@ -52,8 +52,9 @@ def call_handler_example(): 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 events it yields or returns are -queued after the current event, as if the caller had yielded them. +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): @@ -94,7 +95,8 @@ state. - 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; they run one after the other on the same state. +- 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 diff --git a/news/+awaitable-event-handlers.breaking.md b/news/+awaitable-event-handlers.breaking.md new file mode 100644 index 00000000000..4489250c37f --- /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` raises `TypeError` when the state class is created: apply other decorators below `@rx.event`. 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 index 47b4584ea77..a15586d2bae 100644 --- a/news/+awaitable-event-handlers.feature.md +++ b/news/+awaitable-event-handlers.feature.md @@ -1,4 +1,4 @@ -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, and a handler may now return any value rather than only events. `@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. +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 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..4489250c37f --- /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` raises `TypeError` when the state class is created: apply other decorators below `@rx.event`. 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 index 47b4584ea77..a15586d2bae 100644 --- a/packages/reflex-base/news/+awaitable-event-handlers.feature.md +++ b/packages/reflex-base/news/+awaitable-event-handlers.feature.md @@ -1,4 +1,4 @@ -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, and a handler may now return any value rather than only events. `@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. +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 diff --git a/packages/reflex-base/src/reflex_base/event/__init__.py b/packages/reflex-base/src/reflex_base/event/__init__.py index 40378942b62..8e4342f7790 100644 --- a/packages/reflex-base/src/reflex_base/event/__init__.py +++ b/packages/reflex-base/src/reflex_base/event/__init__.py @@ -3424,7 +3424,7 @@ def wrapper(func: Callable[..., Any]) -> EventCallback: ) object.__setattr__(func, "__name__", name) object.__setattr__(func, "__qualname__", name) - if state_cls and issubclass(state_cls, BaseState): + if isinstance(state_cls, type) and issubclass(state_cls, BaseState): state_cls._add_event_handler(name, func) return getattr(state_cls, name) return RegistrationContext.register_event_handler( diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index d1980488167..41839e88680 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -2,11 +2,12 @@ from __future__ import annotations +import asyncio import contextlib import dataclasses import functools import os -from collections.abc import Callable, Iterator, Mapping, Sequence +from collections.abc import AsyncIterator, Callable, Mapping, Sequence from typing import TYPE_CHECKING, Any, Protocol from reflex_base import otel @@ -123,6 +124,12 @@ class EventContext(BaseContext): 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) @@ -162,8 +169,25 @@ def set_locked_root(self, root_state: BaseState | None) -> None: """ object.__setattr__(self, "locked_root", root_state) - @contextlib.contextmanager - def holding_lock(self, root_state: BaseState) -> Iterator[None]: + 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. + """ + current = asyncio.current_task() + pending = [t for t in self.inline_tasks if t is not current and not t.done()] + if pending: + if current is not None and current.cancelling(): + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + self.set_locked_root(None) + + @contextlib.asynccontextmanager + async def holding_lock(self, root_state: BaseState) -> AsyncIterator[None]: """Mark the state lock as held for the duration of the block. Args: @@ -176,7 +200,7 @@ def holding_lock(self, root_state: BaseState) -> Iterator[None]: try: yield finally: - self.set_locked_root(None) + await self.release_lock() async def emit_delta(self, delta: Mapping[str, Mapping[str, Any]]) -> None: """Emit a delta to the frontend. 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 88c7a7237ee..06f40bfd84b 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 @@ -381,10 +382,23 @@ async def execute_event_inline(spec: EventSpec) -> Any: "state lock is held, as it would wait for that lock; yield it instead." ) raise EventHandlerTypeError(msg) - substate = await root_state.get_state(registered.states[0]) - return await process_event( - handler=handler, payload=event.payload, state=substate, root_state=root_state - ) + # 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. + task = asyncio.current_task() + if task is not None: + ctx.inline_tasks.add(task) + try: + substate = await root_state.get_state(registered.states[0]) + 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( @@ -700,7 +714,7 @@ async def _execute_event( substate = await state.get_state(registered_handler.states[0]) root_state = state._get_root_state() - with ctx.holding_lock(root_state): + async with ctx.holding_lock(root_state): if needs_to_rehydrate: await self._rehydrate(root_state) 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 8f04458c097..3336b57656f 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 @@ -18,6 +18,7 @@ from reflex.app_mixins.middleware import MiddlewareMixin from reflex.istate.manager import StateManager from reflex_base import otel +from reflex_base.event import EventHandler, EventSpec from reflex_base.event.context import EventContext from reflex_base.event.processor.future import EventFuture from reflex_base.event.processor.timeout import DrainTimeoutManager @@ -27,7 +28,7 @@ if TYPE_CHECKING: from reflex.app import EventNamespace - from reflex.event import Event, EventSpec + from reflex.event import Event if hasattr(asyncio, "QueueShutDown"): @@ -735,9 +736,11 @@ async def _execute_event( """ event = entry.event result = registered_handler.handler.fn(**event.payload) - # Only a coroutine is run for the handler: an awaitable EventSpec it - # returns is a description of an event, not work to do here. - if inspect.iscoroutine(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 diff --git a/pyproject.toml b/pyproject.toml index 8d59db5a773..1eef5b8d253 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "starlette >=1.3.1", "typing_extensions >=4.13.0", "wrapt >=1.17.0,<2.6", - "reflex-base >= 0.10.1", + "reflex-base >= 0.10.2.dev0", "reflex-components-code >= 0.10.1", "reflex-components-core >= 0.10.0", "reflex-components-dataeditor >= 0.10.1", diff --git a/reflex/app.py b/reflex/app.py index 9c37e9df036..2fd832a8e26 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -1899,7 +1899,7 @@ class cannot be found. # No other event handler can modify the state while in this context. if forked_context is not None: # An awaited handler runs inline on the locked tree. - with forked_context.holding_lock(state._get_root_state()): + async with forked_context.holding_lock(state._get_root_state()): yield state else: yield state diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py index b023fa43701..682489019af 100644 --- a/reflex/istate/proxy.py +++ b/reflex/istate/proxy.py @@ -327,7 +327,7 @@ async def __aexit__(self, *exc_info: Any) -> None: finally: try: if self._self_mutable and self._self_actx is not None: - EventContext.get().set_locked_root(None) + await EventContext.get().release_lock() await self._self_actx.__aexit__(*exc_info) finally: self._self_actx = 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 ea24a219f3d..36347935b97 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 @@ -4132,3 +4132,112 @@ async def bg_caller(self): BaseStateToken(ident=token, cls=State) ) assert (await state.get_state(StatelessCallerState)).got == 10 + + +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"] 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 1eae8ffe638..f326a4724be 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -1817,3 +1817,26 @@ async def _returns_spec() -> EventSpec: # noqa: RUF029 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/test_event.py b/tests/units/test_event.py index a3c6d7ee7a9..d73dc45d071 100644 --- a/tests/units/test_event.py +++ b/tests/units/test_event.py @@ -1754,3 +1754,18 @@ def ping(x: str) -> str: 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 == () From 80e1d44c3c62cef134d51954048f2e186167a39b Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:42:52 +0000 Subject: [PATCH 06/13] Release the state lock when the wait for inline runs is cancelled A background handler leaving `async with self` waits for detached inline runs; cancelling it during that wait must still exit the state manager's lock, and the pending inline runs are cancelled rather than left behind. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/context.py | 16 +++-- reflex/istate/proxy.py | 8 ++- .../processor/test_base_state_processor.py | 58 +++++++++++++++++++ 3 files changed, 74 insertions(+), 8 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index 41839e88680..f4415dfea5d 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -179,12 +179,16 @@ async def release_lock(self) -> None: """ current = asyncio.current_task() pending = [t for t in self.inline_tasks if t is not current and not t.done()] - if pending: - if current is not None and current.cancelling(): - for task in pending: - task.cancel() - await asyncio.gather(*pending, return_exceptions=True) - self.set_locked_root(None) + try: + if pending: + try: + await asyncio.gather(*pending, return_exceptions=True) + except asyncio.CancelledError: + for task in pending: + task.cancel() + raise + finally: + self.set_locked_root(None) @contextlib.asynccontextmanager async def holding_lock(self, root_state: BaseState) -> AsyncIterator[None]: diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py index 682489019af..0f37bcd5acb 100644 --- a/reflex/istate/proxy.py +++ b/reflex/istate/proxy.py @@ -327,8 +327,12 @@ async def __aexit__(self, *exc_info: Any) -> None: finally: try: if self._self_mutable and self._self_actx is not None: - await EventContext.get().release_lock() - 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 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 36347935b97..ceea0a2a126 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 @@ -4241,3 +4241,61 @@ def next_event(self): 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"] From ad34033113ec51c881c10929d43a54f9a9cce30c Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:47:52 +0000 Subject: [PATCH 07/13] Drain late and nested inline runs before releasing the state lock A nested inline run on an already registered task no longer drops the outer run's registration, the release keeps draining runs started while it waits, and exceptions of gathered runs are logged instead of being retrieved silently. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/context.py | 23 ++++- .../event/processor/base_state_processor.py | 7 +- .../processor/test_base_state_processor.py | 86 +++++++++++++++++++ 3 files changed, 111 insertions(+), 5 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index f4415dfea5d..ab49ae66c4f 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -6,6 +6,7 @@ import contextlib import dataclasses import functools +import logging import os from collections.abc import AsyncIterator, Callable, Mapping, Sequence from typing import TYPE_CHECKING, Any, Protocol @@ -23,6 +24,9 @@ from reflex_base.event.processor.future import EventFuture +logger = logging.getLogger(__name__) + + @functools.lru_cache def get_name(cls: type | Callable) -> str: """Get the name of the state/func. @@ -178,15 +182,28 @@ async def release_lock(self) -> None: cancelled, those runs are cancelled too. """ current = asyncio.current_task() - pending = [t for t in self.inline_tasks if t is not current and not t.done()] try: - if pending: + # 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() + ]: try: - await asyncio.gather(*pending, return_exceptions=True) + 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) 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 06f40bfd84b..450c57c32ba 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 @@ -385,8 +385,11 @@ async def execute_event_inline(spec: EventSpec) -> Any: # 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 not None: + added = task is not None and task not in ctx.inline_tasks + if added: ctx.inline_tasks.add(task) try: substate = await root_state.get_state(registered.states[0]) @@ -397,7 +400,7 @@ async def execute_event_inline(spec: EventSpec) -> Any: root_state=root_state, ) finally: - if task is not None: + if added: ctx.inline_tasks.discard(task) 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 ceea0a2a126..0810a75e35d 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 @@ -4299,3 +4299,89 @@ def ping(self): 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 From ff91379a1423a52f12aa4f0922a25d0de3c36a18 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:51:11 +0000 Subject: [PATCH 08/13] Cancel inline runs for an already cancelled handler and trim the lock hot path A task cancelled inside the locked block is not interrupted again while it waits for detached inline runs, so those are cancelled up front. The per-event lock marking uses a plain try/finally and a fast path when nothing ran inline. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/context.py | 11 ++++ .../event/processor/base_state_processor.py | 7 ++- .../processor/test_base_state_processor.py | 59 +++++++++++++++++++ 3 files changed, 76 insertions(+), 1 deletion(-) diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index ab49ae66c4f..6e9d05f3da4 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -181,13 +181,24 @@ async def release_lock(self) -> None: 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: 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 450c57c32ba..9544c5f21ed 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 @@ -717,7 +717,10 @@ async def _execute_event( substate = await state.get_state(registered_handler.states[0]) root_state = state._get_root_state() - async with ctx.holding_lock(root_state): + # 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) @@ -729,6 +732,8 @@ async def _execute_event( 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) 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 0810a75e35d..9b5a4f32b60 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 @@ -4385,3 +4385,62 @@ def next_event(self): "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"] From f87f474d9b78ae2655fe7d413286182248a15263 Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:52:55 +0000 Subject: [PATCH 09/13] Narrow the inline task registration for pyright The CI pyright does not carry the `added` narrowing into `set.add`; keep the task only when it was registered here. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../reflex_base/event/processor/base_state_processor.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 9544c5f21ed..591cbf2f3a4 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 @@ -388,8 +388,10 @@ async def execute_event_inline(spec: EventSpec) -> Any: # A nested inline run on an already registered task leaves the outer # run's registration in place. task = asyncio.current_task() - added = task is not None and task not in ctx.inline_tasks - if added: + 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]) @@ -400,7 +402,7 @@ async def execute_event_inline(spec: EventSpec) -> Any: root_state=root_state, ) finally: - if added: + if task is not None: ctx.inline_tasks.discard(task) From 42c7239af2ab9624b04b441d6ab9e2a50b12d53d Mon Sep 17 00:00:00 2001 From: Farhan Date: Tue, 27 Oct 2026 17:56:16 +0000 Subject: [PATCH 10/13] Mark the lock with a plain try/finally in modify_state too Drops the unused holding_lock context manager: both callers now set and release the lock marker directly, which keeps the per-call overhead down on the shared-state fan-out path. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../src/reflex_base/event/context.py | 19 +------------------ reflex/app.py | 14 +++++++++----- 2 files changed, 10 insertions(+), 23 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index 6e9d05f3da4..af4585b3d83 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -3,12 +3,11 @@ from __future__ import annotations import asyncio -import contextlib import dataclasses import functools import logging import os -from collections.abc import AsyncIterator, Callable, Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from typing import TYPE_CHECKING, Any, Protocol from reflex_base import otel @@ -218,22 +217,6 @@ async def release_lock(self) -> None: finally: self.set_locked_root(None) - @contextlib.asynccontextmanager - async def holding_lock(self, root_state: BaseState) -> AsyncIterator[None]: - """Mark the state lock as held for the duration of the block. - - Args: - root_state: The root of the locked state tree. - - Yields: - Nothing. - """ - self.set_locked_root(root_state) - try: - yield - finally: - await self.release_lock() - async def emit_delta(self, delta: Mapping[str, Mapping[str, Any]]) -> None: """Emit a delta to the frontend. diff --git a/reflex/app.py b/reflex/app.py index 2fd832a8e26..be97f853734 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -1897,12 +1897,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. - if forked_context is not None: - # An awaited handler runs inline on the locked tree. - async with forked_context.holding_lock(state._get_root_state()): - yield state - else: + 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: From 2b6f38fa94e7b6f3e1304a5b3a8d2df133cdb6fe Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 12:37:39 +0000 Subject: [PATCH 11/13] Do not cache literal var dispatch by a proxy's type _literal_var_for caches the literal class per exact value type, but a MutableProxy has the proxy's type whatever it wraps while isinstance sees the wrapped value. The first proxied value pinned every later proxy to its literal class, so LiteralVar.create(proxy_of_list) raised after LiteralVar.create(proxy_of_dict). Skip the cache when the value reports a __class__ other than its type. Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- .../news/+literal-var-proxy-cache.bugfix.md | 1 + .../reflex-base/src/reflex_base/vars/base.py | 6 +++-- tests/units/reflex_base/vars/test_base.py | 23 +++++++++++++++++++ 3 files changed, 28 insertions(+), 2 deletions(-) create mode 100644 packages/reflex-base/news/+literal-var-proxy-cache.bugfix.md 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/vars/base.py b/packages/reflex-base/src/reflex_base/vars/base.py index c010903b219..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 diff --git a/tests/units/reflex_base/vars/test_base.py b/tests/units/reflex_base/vars/test_base.py index 242cf2e85a9..6abb2d257b5 100644 --- a/tests/units/reflex_base/vars/test_base.py +++ b/tests/units/reflex_base/vars/test_base.py @@ -564,6 +564,29 @@ def create(cls, value: Coordinate, _var_data=None): base._literal_var_by_type.clear() +def test_literal_var_dispatch_sees_through_proxies(): + """Proxies of different values dispatch on the wrapped value, not the proxy type.""" + 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.""" From db34f6432c959a5cad16f455a1a8873cd1fd8131 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 12:47:56 +0000 Subject: [PATCH 12/13] Isolate the proxy dispatch test's state class in a clean registry Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- tests/units/reflex_base/vars/test_base.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/units/reflex_base/vars/test_base.py b/tests/units/reflex_base/vars/test_base.py index 6abb2d257b5..c9045d71638 100644 --- a/tests/units/reflex_base/vars/test_base.py +++ b/tests/units/reflex_base/vars/test_base.py @@ -564,8 +564,12 @@ def create(cls, value: Coordinate, _var_data=None): base._literal_var_by_type.clear() -def test_literal_var_dispatch_sees_through_proxies(): - """Proxies of different values dispatch on the wrapped value, not the proxy type.""" +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( From 2dff931aa0b9c660f31318a96ec6f8f9f6cafde3 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 12:54:46 +0000 Subject: [PATCH 13/13] Fetch the quote outside the state lock in the stateless handler example Claude-Session: https://claude.ai/code/session_01LwJUCVfbdzZayfSZMiP8GJ --- docs/events/decentralized_event_handlers.md | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/docs/events/decentralized_event_handlers.md b/docs/events/decentralized_event_handlers.md index a3799bafa4b..7f7df4560a9 100644 --- a/docs/events/decentralized_event_handlers.md +++ b/docs/events/decentralized_event_handlers.md @@ -177,10 +177,17 @@ async def fetch_quote(symbol: str) -> float: class PortfolioState(rx.State): price: float = 0.0 - @rx.event + @rx.event(background=True) async def refresh(self, symbol: str): - self.price = await fetch_quote(symbol) + price = await fetch_quote(symbol) + async with self: + self.price = price ``` +Call external services from a [background task](/docs/events/background-events/) +so the client's state lock is not held while waiting: a stateless handler runs +inline wherever it is awaited, so the quote above is fetched outside +`async with self` and the lock is taken only to store it. + A stateless handler can still yield or return events, which run against the state of the client that triggered it.