From 517c5382460c8b092533918be006cbd1cc2e2208 Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 12:32:52 -0700
Subject: [PATCH 01/11] fix(events): stop background tasks from cleaning the
 root state unlocked

After dropping the state lock, the background branch of _execute_event
passed root_state into process_event, whose trailing chain_updates
snapshotted dirty vars, suspended for delta resolution and the socket
emit, and then ran _clean() on the root, all without holding the lock.

On a shared state tree (opportunistic locking, or the in-memory state
manager) that window races concurrent events: a foreground handler's
write landing between the snapshot and the _clean() is discarded before
any delta harvests it, so the value never reaches the frontend. With a
1 Hz background poller this silently eats roughly 1 in 8 foreground
updates that land near a poll tick.

Pass root_state=None instead: the trailing chain_updates still routes
yielded and returned events, but does no delta work. A background
task's own state changes are emitted and cleaned by its async-with-self
context exits, which re-acquire the lock, so nothing is lost.
---
 packages/reflex-base/news/6920.bugfix.md      |   1 +
 .../event/processor/base_state_processor.py   |  20 +++-
 .../processor/test_base_state_processor.py    | 102 ++++++++++++++++++
 3 files changed, 119 insertions(+), 4 deletions(-)
 create mode 100644 packages/reflex-base/news/6920.bugfix.md

diff --git a/packages/reflex-base/news/6920.bugfix.md b/packages/reflex-base/news/6920.bugfix.md
new file mode 100644
index 00000000000..c446b5d6b35
--- /dev/null
+++ b/packages/reflex-base/news/6920.bugfix.md
@@ -0,0 +1 @@
+Stop background event handlers from computing a delta and cleaning the root state after the state lock is dropped. On a shared state tree (opportunistic locking, in-memory state manager) that unlocked snapshot-then-clean raced concurrent foreground handlers: a write landing between the background task's dirty-var snapshot and its `_clean()` was silently discarded and never reached any delta. Background state changes are emitted by their `async with self` context exits, which hold the lock.
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 4185cef9e99..307686e7d63 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
@@ -232,7 +232,7 @@ async def process_event(
     handler: EventHandler,
     payload: dict,
     state: BaseState | StateProxy,
-    root_state: BaseState,
+    root_state: BaseState | None,
 ):
     """Process event.
 
@@ -240,7 +240,12 @@ async def process_event(
         handler: EventHandler to process.
         payload: The event payload.
         state: State to process the handler.
-        root_state: The root state of the app, used for emitting deltas.
+        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
+            concurrent events on a shared state tree, and background state
+            changes are emitted by the ``async with self`` context exits
+            instead.
 
     Raises:
         ValueError: If a string value is received for an int or float type and cannot be converted.
@@ -393,12 +398,19 @@ async def _execute_event(
                     root_state=root_state,
                 )
                 return
-        # Otherwise drop the state lock and start processing the background task with a proxy 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)
+        # computing a delta here races whatever event holds the lock now -- a
+        # foreground write landing between this task's dirty-var snapshot and
+        # its _clean() would be discarded before any delta carries it. A
+        # background task's own state changes are emitted (and cleaned) by its
+        # `async with self` context exits, which re-acquire the lock.
         await process_event(
             handler=registered_handler.handler,
             state=StateProxy(substate),
             payload=event.payload,
-            root_state=root_state,
+            root_state=None,
         )
 
     async def _handle_backend_exception(
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 6f89d9adf49..0a6d4b5d8d2 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
@@ -17,6 +17,7 @@
 from reflex.app import App
 from reflex.event import Event
 from reflex.istate.manager.memory import StateManagerMemory
+from reflex.istate.manager.token import BaseStateToken
 from reflex.middleware.middleware import Middleware
 from reflex.state import OnLoadInternalState, State, StateUpdate
 
@@ -214,3 +215,104 @@ async def preprocess(self, app, state, event) -> StateUpdate:
     client_event_names = {e.name for _, events in emitted_events for e in events}
     assert "_call_function" in client_event_names
     assert "_redirect" in client_event_names
+
+
+async def test_background_event_does_not_discard_concurrent_foreground_write(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """A foreground write racing a background task's completion reaches a delta.
+
+    Regression: after dropping the state lock, the background branch passed
+    ``root_state`` into ``process_event``, whose trailing ``chain_updates``
+    snapshotted dirty vars, suspended (delta resolution / emit), and then ran
+    ``_clean()`` -- all unlocked. On a shared state tree (opportunistic
+    locking, or the in-memory manager used here) a foreground handler's write
+    landing inside that snapshot->clean window was cleaned before any delta
+    harvested it: the value never reached the frontend.
+
+    The gates below force exactly that interleaving: the background task's
+    trailing delta resolution (via the uncached async computed var) signals
+    the foreground handler to write, then resumes -- pre-fix its ``_clean()``
+    discarded the write mid-foreground-handler.
+
+    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.
+    """
+    import asyncio
+    import contextlib
+
+    bg_started = asyncio.Event()
+    bg_resolving = asyncio.Event()
+    fg_wrote = asyncio.Event()
+    hold_resolution = [False]
+
+    class BgRaceState(State):
+        victim: str = ""
+
+        @rx.var(cache=False)
+        async def window(self) -> int:
+            # Uncached: recomputed in every delta. Once armed, the background
+            # task's trailing delta resolution parks here until the foreground
+            # handler has written -- holding the snapshot->clean window open.
+            if hold_resolution[0]:
+                bg_resolving.set()
+                await fg_wrote.wait()
+            return 0
+
+        @event(background=True)
+        async def bg(self):
+            hold_resolution[0] = True
+            bg_started.set()
+
+        @event
+        async def fg(self):
+            # Bounded: with the fix, the background task does no trailing
+            # delta work, so nothing ever sets bg_resolving.
+            with contextlib.suppress(TimeoutError):
+                await asyncio.wait_for(bg_resolving.wait(), timeout=0.5)
+            self.victim = "written"
+            fg_wrote.set()
+            # Yield so the background task's (pre-fix) clean can land while
+            # this handler is still mid-flight, as a real await would allow.
+            for _ in range(20):
+                await asyncio.sleep(0)
+
+    # Seed router_data up front so no event triggers _rehydrate (whose
+    # full-dict resolution would park on the armed gate before the foreground
+    # handler could run) and no event needs to carry router_data of its own.
+    assert real_base_state_processor._root_context is not None
+    state_manager = real_base_state_processor._root_context.state_manager
+    async with state_manager.modify_state(
+        BaseStateToken(ident=token, cls=State)
+    ) as seed_root:
+        seed_root.router_data = {"pathname": "/", "query": {}}
+    try:
+        async with real_base_state_processor as processor:
+            await processor.enqueue(token, Event.from_event_type(BgRaceState.bg())[0])
+            await asyncio.wait_for(bg_started.wait(), timeout=2)
+            await processor.enqueue(token, Event.from_event_type(BgRaceState.fg())[0])
+            await processor.join(5)
+    finally:
+        # The uncached computed var registered BgRaceState as an always-dirty
+        # substate on the shared State class; later tests' state trees don't
+        # contain it and would KeyError in get_delta (see reload_state_module).
+        State._always_dirty_substates.discard(BgRaceState.get_name())
+
+    state_name = BgRaceState.get_full_name()
+    victim_key = "victim" + FIELD_MARKER
+    delivered = [
+        d[state_name][victim_key]
+        for _, d in emitted_deltas
+        if state_name in d and d[state_name].get(victim_key) == "written"
+    ]
+    assert delivered, (
+        "The foreground handler's write never reached any delta; it was "
+        "cleaned by the background task's unlocked trailing update. Deltas: "
+        f"{emitted_deltas}"
+    )

From a4a23ea5e8648b1a16183a2134e2233b2c098b4a Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 12:43:52 -0700
Subject: [PATCH 02/11] fix(events): flush yield deltas only while the state
 lock is held

Two follow-ups from review and CI:

Background generators yielding from inside async-with-self are suspended
while their proxy still holds the lock, so flushing the delta there is
safe and keeps the documented ordering (deltas reach the frontend before
the yielded event). Gate the per-yield flush on the lock being held
instead of dropping it for the whole background handler.

The regression test's wait_for suppress missed on py3.10, where
asyncio.TimeoutError is not yet the builtin; use asyncio.wait, which
does not raise on timeout.
---
 .../event/processor/base_state_processor.py   | 41 +++++++--
 .../processor/test_base_state_processor.py    | 90 ++++++++++++++++++-
 2 files changed, 121 insertions(+), 10 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 307686e7d63..7e9ba822fad 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
@@ -264,6 +264,25 @@ async def process_event(
             f"Error transforming event payload for handler {handler_name}: {ex}"
         )
 
+    def _flush_root_state() -> BaseState | None:
+        """The root to flush deltas from, only while the state lock is held.
+
+        Foreground handlers pass the locked root in. A background handler
+        yielding from inside ``async with self`` is suspended while its proxy
+        still holds the lock, so flushing through the proxy's root keeps the
+        documented ordering: deltas reach the frontend before the yielded
+        event. Outside the proxy context there is no lock, and flushing there
+        is the unlocked snapshot/clean that discards concurrent writes.
+
+        Returns:
+            The root state to flush, or None when the lock is not held.
+        """
+        if root_state is not None:
+            return root_state
+        if isinstance(state, StateProxy) and state._is_mutable():
+            return state.__wrapped__._get_root_state()
+        return None
+
     # Handle async functions.
     if inspect.iscoroutinefunction(fn.func):
         events = await fn(**payload)
@@ -274,28 +293,38 @@ async def process_event(
     # Handle async generators.
     if inspect.isasyncgen(events):
         async for event in events:
-            await chain_updates(event, root_state=root_state, handler_name=handler_name)
-        await chain_updates(None, root_state=root_state, handler_name=handler_name)
+            await chain_updates(
+                event, root_state=_flush_root_state(), handler_name=handler_name
+            )
+        await chain_updates(
+            None, root_state=_flush_root_state(), handler_name=handler_name
+        )
 
     # Handle regular generators.
     elif inspect.isgenerator(events):
         try:
             while True:
                 await chain_updates(
-                    next(events), root_state=root_state, handler_name=handler_name
+                    next(events),
+                    root_state=_flush_root_state(),
+                    handler_name=handler_name,
                 )
         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=root_state, handler_name=handler_name
+                    si.value, root_state=_flush_root_state(), handler_name=handler_name
                 )
-        await chain_updates(None, root_state=root_state, handler_name=handler_name)
+        await chain_updates(
+            None, root_state=_flush_root_state(), handler_name=handler_name
+        )
 
     # Handle regular event chains.
     else:
-        await chain_updates(events, root_state=root_state, handler_name=handler_name)
+        await chain_updates(
+            events, root_state=_flush_root_state(), handler_name=handler_name
+        )
 
 
 class BaseStateEventProcessor(EventProcessor):
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 0a6d4b5d8d2..02a13417c5d 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
@@ -245,7 +245,6 @@ async def test_background_event_does_not_discard_concurrent_foreground_write(
         token: The client token.
     """
     import asyncio
-    import contextlib
 
     bg_started = asyncio.Event()
     bg_resolving = asyncio.Event()
@@ -273,9 +272,12 @@ async def bg(self):
         @event
         async def fg(self):
             # Bounded: with the fix, the background task does no trailing
-            # delta work, so nothing ever sets bg_resolving.
-            with contextlib.suppress(TimeoutError):
-                await asyncio.wait_for(bg_resolving.wait(), timeout=0.5)
+            # delta work, so nothing ever sets bg_resolving. asyncio.wait
+            # instead of wait_for: on py3.10 the latter raises
+            # asyncio.TimeoutError, which is not yet the builtin.
+            waiter = asyncio.ensure_future(bg_resolving.wait())
+            await asyncio.wait([waiter], timeout=0.5)
+            waiter.cancel()
             self.victim = "written"
             fg_wrote.set()
             # Yield so the background task's (pre-fix) clean can land while
@@ -316,3 +318,83 @@ async def fg(self):
         "cleaned by the background task's unlocked trailing update. Deltas: "
         f"{emitted_deltas}"
     )
+
+
+async def test_background_yield_inside_context_flushes_delta_before_event(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A yield inside ``async with self`` emits the delta before the event.
+
+    A background generator suspended at a yield inside its proxy context
+    still holds the state lock, so flushing the delta there is safe and
+    preserves the documented ordering: frontend events are processed with
+    the latest state. Only yields outside the context (no lock) must skip
+    the flush.
+
+    Args:
+        wired_app: The App wired to the processor's state manager.
+        real_base_state_processor: The unmocked BaseStateEventProcessor.
+        token: The client token.
+    """
+    import asyncio
+
+    timeline: list[tuple[str, Any]] = []
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+
+    async def record_delta(tok: str, delta: Mapping[str, Mapping[str, Any]]) -> None:  # noqa: RUF029
+        timeline.append(("delta", delta))
+
+    async def record_event(tok: str, *events: Event) -> None:  # noqa: RUF029
+        timeline.append(("event", tuple(ev.name for ev in events)))
+
+    object.__setattr__(root_ctx, "emit_delta_impl", record_delta)
+    object.__setattr__(root_ctx, "emit_event_impl", record_event)
+
+    class BgYieldOrderState(State):
+        marker: str = ""
+
+        @event(background=True)
+        async def bg_yield(self):
+            async with self:
+                self.marker = "set"
+                yield rx.call_script("void 0")
+
+    state_manager = root_ctx.state_manager
+    async with state_manager.modify_state(
+        BaseStateToken(ident=token, cls=State)
+    ) as seed_root:
+        seed_root.router_data = {"pathname": "/", "query": {}}
+
+    async with real_base_state_processor as processor:
+        await processor.enqueue(
+            token, Event.from_event_type(BgYieldOrderState.bg_yield())[0]
+        )
+        await processor.join(5)
+
+    state_name = BgYieldOrderState.get_full_name()
+    marker_key = "marker" + FIELD_MARKER
+    delta_index = next(
+        (
+            index
+            for index, (kind, payload) in enumerate(timeline)
+            if kind == "delta" and payload.get(state_name, {}).get(marker_key) == "set"
+        ),
+        None,
+    )
+    event_index = next(
+        (
+            index
+            for index, (kind, payload) in enumerate(timeline)
+            if kind == "event" and "_call_script" in payload
+        ),
+        None,
+    )
+    assert delta_index is not None, f"marker delta never emitted: {timeline}"
+    assert event_index is not None, f"yielded event never emitted: {timeline}"
+    assert delta_index < event_index, (
+        "The yielded frontend event was emitted before the delta for the "
+        f"mutation made in the same proxy context: {timeline}"
+    )

From 4a623791a60ed5d0ff385b3dc79f23a9aa1311df Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 12:46:07 -0700
Subject: [PATCH 03/11] chore: drop unused import flagged by ruff

---
 .../reflex_base/event/processor/test_base_state_processor.py    | 2 --
 1 file changed, 2 deletions(-)

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 02a13417c5d..e0a736405a3 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
@@ -338,8 +338,6 @@ async def test_background_yield_inside_context_flushes_delta_before_event(
         real_base_state_processor: The unmocked BaseStateEventProcessor.
         token: The client token.
     """
-    import asyncio
-
     timeline: list[tuple[str, Any]] = []
     root_ctx = real_base_state_processor._root_context
     assert root_ctx is not None

From 83f60e640f1a3a685d4685a0fcd853802b352516 Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 13:04:40 -0700
Subject: [PATCH 04/11] test: drive the regression interleaving with gates
 instead of timeouts

The foreground handler now proceeds on FIRST_COMPLETED of the trailing
resolution's signal (pre-fix code) or the background event's own future
(fixed code), and afterwards awaits that future, which completes
strictly after the trailing clean. No wall-clock waits remain in either
world, and the pre-fix failure is forced by construction rather than by
a tick budget.
---
 .../processor/test_base_state_processor.py    | 42 ++++++++++++-------
 1 file changed, 27 insertions(+), 15 deletions(-)

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 e0a736405a3..49535fbb28d 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
@@ -233,10 +233,12 @@ async def test_background_event_does_not_discard_concurrent_foreground_write(
     landing inside that snapshot->clean window was cleaned before any delta
     harvested it: the value never reached the frontend.
 
-    The gates below force exactly that interleaving: the background task's
-    trailing delta resolution (via the uncached async computed var) signals
-    the foreground handler to write, then resumes -- pre-fix its ``_clean()``
-    discarded the write mid-foreground-handler.
+    Every step is gate-driven, in both worlds: pre-fix, the background
+    task's trailing delta resolution (via the uncached async computed var)
+    parks until the foreground handler has written, and the background
+    event's future completes strictly after its trailing ``_clean()``; on
+    fixed code that future completes without any trailing resolution and
+    the foreground handler proceeds immediately.
 
     Args:
         wired_app: The App wired to the processor's state manager.
@@ -250,6 +252,7 @@ async def test_background_event_does_not_discard_concurrent_foreground_write(
     bg_resolving = asyncio.Event()
     fg_wrote = asyncio.Event()
     hold_resolution = [False]
+    bg_future_box: list = []
 
     class BgRaceState(State):
         victim: str = ""
@@ -271,19 +274,21 @@ async def bg(self):
 
         @event
         async def fg(self):
-            # Bounded: with the fix, the background task does no trailing
-            # delta work, so nothing ever sets bg_resolving. asyncio.wait
-            # instead of wait_for: on py3.10 the latter raises
-            # asyncio.TimeoutError, which is not yet the builtin.
+            # Proceed once the background trailing resolution has taken its
+            # snapshot (pre-fix code parks it on fg_wrote), or once the
+            # background event has fully completed without one (fixed code).
             waiter = asyncio.ensure_future(bg_resolving.wait())
-            await asyncio.wait([waiter], timeout=0.5)
+            await asyncio.wait(
+                [waiter, *bg_future_box], return_when=asyncio.FIRST_COMPLETED
+            )
             waiter.cancel()
             self.victim = "written"
             fg_wrote.set()
-            # Yield so the background task's (pre-fix) clean can land while
-            # this handler is still mid-flight, as a real await would allow.
-            for _ in range(20):
-                await asyncio.sleep(0)
+            # Pre-fix, the parked resolution now resumes, emits, and cleans;
+            # the background event's future completes strictly after that
+            # clean, so awaiting it guarantees the clean landed before this
+            # handler's own delta snapshot. On fixed code it is already done.
+            await asyncio.wait(bg_future_box)
 
     # Seed router_data up front so no event triggers _rehydrate (whose
     # full-dict resolution would park on the armed gate before the foreground
@@ -296,8 +301,15 @@ async def fg(self):
         seed_root.router_data = {"pathname": "/", "query": {}}
     try:
         async with real_base_state_processor as processor:
-            await processor.enqueue(token, Event.from_event_type(BgRaceState.bg())[0])
-            await asyncio.wait_for(bg_started.wait(), timeout=2)
+            bg_future_box.append(
+                await processor.enqueue(
+                    token, Event.from_event_type(BgRaceState.bg())[0]
+                )
+            )
+            started = asyncio.ensure_future(bg_started.wait())
+            await asyncio.wait([started], timeout=2)
+            started.cancel()
+            assert bg_started.is_set(), "background handler never started"
             await processor.enqueue(token, Event.from_event_type(BgRaceState.fg())[0])
             await processor.join(5)
     finally:

From 3c52ea3784dd393df594ef946b283a765e7e54be Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 13:30:12 -0700
Subject: [PATCH 05/11] fix(events): make chain_updates atomic and keep the
 background flush, locked

Review follow-ups on the unlocked-clean fix:

chain_updates now snapshots and cleans in one step before awaiting delta
resolution, so a concurrent write landing mid-resolution stays dirty for
the next harvest even if a caller ever passes an unlocked root -- the
protection lives in the primitive, not only in the caller's gate.

A background handler that never enters async-with-self used to get its
only delta from the removed trailing flush (uncached computed vars
refreshed on every background tick). Restore that flush, under the
state lock, detected via a proxy subclass that records context entry.

Also move a function-level asyncio import to the module top.
---
 .../event/processor/base_state_processor.py   |  56 +++++++-
 .../processor/test_base_state_processor.py    | 129 +++++++++++++++++-
 2 files changed, 178 insertions(+), 7 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 7e9ba822fad..19e715a5838 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
@@ -209,17 +209,22 @@ async def chain_updates(
         root_state: The root state of the app, no delta emitted if omitted.
     """
     from reflex.event import Event
+    from reflex.state import _resolve_delta
 
     ctx = EventContext.get()
 
     if root_state is not None:
-        # Emit deltas first, so any frontend events are processed with the latest state.
+        # Snapshot and clean in one step, with no suspension between them: a
+        # concurrent write landing while the delta is being resolved or
+        # emitted must stay dirty for the next harvest, not be discarded by a
+        # clean that never snapshotted it. The delta is still emitted before
+        # the events below, so frontend events observe the latest state.
         try:
-            delta = await root_state._get_resolved_delta()
-            if delta:
-                await ctx.emit_delta(delta)
+            delta = root_state.get_delta()
         finally:
             root_state._clean()
+        if delta and (delta := await _resolve_delta(delta)):
+            await ctx.emit_delta(delta)
 
     # Convert valid EventHandler and EventSpec into Event
     if fixed_events := Event.from_event_type(
@@ -327,6 +332,29 @@ def _flush_root_state() -> BaseState | None:
         )
 
 
+class _EnterTrackingStateProxy(StateProxy):
+    """A StateProxy recording whether the handler ever entered ``async with self``."""
+
+    def __init__(self, *args, **kwargs):
+        """Create the proxy with the entered flag unset.
+
+        Args:
+            *args: Positional arguments for StateProxy.
+            **kwargs: Keyword arguments for StateProxy.
+        """
+        super().__init__(*args, **kwargs)
+        self._self_entered_context = False
+
+    async def __aenter__(self):
+        """Enter the mutability context, recording that it was entered.
+
+        Returns:
+            This proxy in mutable mode.
+        """
+        self._self_entered_context = True
+        return await super().__aenter__()
+
+
 class BaseStateEventProcessor(EventProcessor):
     """Event processor for BaseState-derived states.
 
@@ -435,12 +463,30 @@ async def _execute_event(
         # its _clean() would be discarded before any delta carries it. A
         # background task's own state changes are emitted (and cleaned) by its
         # `async with self` context exits, which re-acquire the lock.
+        proxy = _EnterTrackingStateProxy(substate)
         await process_event(
             handler=registered_handler.handler,
-            state=StateProxy(substate),
+            state=proxy,
             payload=event.payload,
             root_state=None,
         )
+        if not proxy._self_entered_context:
+            # A handler that never entered `async with self` emitted nothing,
+            # but every background event used to flush a delta (refreshing
+            # uncached computed vars, and any dirty vars the preamble left,
+            # like router_data). Preserve that, under the lock this time.
+            async with ctx.state_manager.modify_state_with_links(
+                BaseStateToken(
+                    ident=ctx.token,
+                    cls=registered_handler.states[0],
+                ),
+                event=event,
+            ) as flush_state:
+                await chain_updates(
+                    None,
+                    root_state=flush_state._get_root_state(),
+                    handler_name=registered_handler.handler.fn.__qualname__,
+                )
 
     async def _handle_backend_exception(
         self, ex: Exception, ev_ctx: EventContext | None = 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 49535fbb28d..fd63c9bbcee 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
@@ -1,5 +1,6 @@
 """Tests for BaseStateEventProcessor, specifically the _rehydrate path."""
 
+import asyncio
 import traceback
 from collections.abc import Mapping
 from typing import Any
@@ -246,8 +247,6 @@ async def test_background_event_does_not_discard_concurrent_foreground_write(
         emitted_deltas: List to capture emitted deltas.
         token: The client token.
     """
-    import asyncio
-
     bg_started = asyncio.Event()
     bg_resolving = asyncio.Event()
     fg_wrote = asyncio.Event()
@@ -269,6 +268,11 @@ async def window(self) -> int:
 
         @event(background=True)
         async def bg(self):
+            # Enter the context once so the never-entered compatibility flush
+            # (which runs under the lock) stays out of this choreography; arm
+            # the gate only afterwards so the context exit resolves unparked.
+            async with self:
+                pass
             hold_resolution[0] = True
             bg_started.set()
 
@@ -408,3 +412,124 @@ async def bg_yield(self):
         "The yielded frontend event was emitted before the delta for the "
         f"mutation made in the same proxy context: {timeline}"
     )
+
+
+async def test_background_event_without_context_still_flushes_a_delta(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """A background handler with no ``async with self`` still flushes a delta.
+
+    Backward compatibility: before the unlocked trailing flush was removed,
+    every background event emitted a delta, which is what refreshed uncached
+    computed vars for apps driving re-renders off a bare background tick.
+    That flush now runs under the state lock instead of disappearing.
+
+    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 NoContextBgState(State):
+        @rx.var(cache=False)
+        def beat(self) -> int:
+            return 7
+
+        @event(background=True)
+        async def bg(self):
+            pass
+
+    assert real_base_state_processor._root_context is not None
+    state_manager = real_base_state_processor._root_context.state_manager
+    async with state_manager.modify_state(
+        BaseStateToken(ident=token, cls=State)
+    ) as seed_root:
+        seed_root.router_data = {"pathname": "/", "query": {}}
+
+    try:
+        async with real_base_state_processor as processor:
+            await processor.enqueue(
+                token, Event.from_event_type(NoContextBgState.bg())[0]
+            )
+            await processor.join(5)
+    finally:
+        State._always_dirty_substates.discard(NoContextBgState.get_name())
+
+    state_name = NoContextBgState.get_full_name()
+    beat_key = "beat" + FIELD_MARKER
+    assert any(d.get(state_name, {}).get(beat_key) == 7 for _, d in emitted_deltas), (
+        f"no delta refreshed the uncached var: {emitted_deltas}"
+    )
+
+
+async def test_chain_updates_keeps_writes_made_during_delta_resolution(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
+    token: str,
+):
+    """chain_updates must not clean a write it never snapshotted.
+
+    The snapshot and the clean happen in one step, before the resolved delta
+    is awaited or emitted, so a concurrent write landing during resolution
+    stays dirty for the next harvest even if a caller ever runs chain_updates
+    without holding the state lock.
+
+    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.
+    """
+    from reflex_base.event.processor.base_state_processor import chain_updates
+
+    resolving = asyncio.Event()
+    release = asyncio.Event()
+    hold_resolution = [False]
+
+    class MidResolveState(State):
+        victim: str = ""
+
+        @rx.var(cache=False)
+        async def window(self) -> int:
+            if hold_resolution[0]:
+                resolving.set()
+                await release.wait()
+            return 0
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    EventContext.set(root_ctx.fork(token=token))
+    state_manager = root_ctx.state_manager
+
+    try:
+        root = await state_manager.get_state(BaseStateToken(ident=token, cls=State))
+        substate = await root.get_state(MidResolveState)
+        root._clean()
+
+        hold_resolution[0] = True
+        flush = asyncio.ensure_future(
+            chain_updates(None, root_state=root, handler_name="unlocked_flush")
+        )
+        await resolving.wait()
+        substate.victim = "written"
+        hold_resolution[0] = False
+        release.set()
+        await flush
+
+        assert "victim" in substate.dirty_vars, (
+            "the write made during delta resolution was cleaned away"
+        )
+        await chain_updates(None, root_state=root, handler_name="second_flush")
+    finally:
+        State._always_dirty_substates.discard(MidResolveState.get_name())
+
+    state_name = MidResolveState.get_full_name()
+    victim_key = "victim" + FIELD_MARKER
+    assert any(
+        d.get(state_name, {}).get(victim_key) == "written" for _, d in emitted_deltas
+    ), f"the surviving write never reached a delta: {emitted_deltas}"

From 636fd7a78002eec88596c0d88dee217a24498e8a Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Thu, 20 Aug 2026 13:56:50 -0700
Subject: [PATCH 06/11] fix(events): revert the atomic snapshot-clean in
 chain_updates

The reorder broke SharedState fan-out (test_linked_state, all memory
legs): resolving the delta is itself what re-marks linked vars dirty,
through the patch machinery that async computed vars reach via
get_state, and SharedState._clean captures its dirty vars for the
cross-client fan-out at clean time. Cleaning before resolution captured
the pre-resolution set, so the seed lost the linked writes and other
clients never saw them.

Post-snapshot dirt from that machinery is indistinguishable from a
concurrent foreground write at clean time, so there is no cheap
selective clean either; hardening the primitive needs the fan-out
capture decoupled from _clean, left as follow-up. The background-branch
gate and the locked trailing flush, which fix the observed lost-update
bug, are unchanged. The chain_updates ordering dependency is now
documented at the clean site.
---
 .../event/processor/base_state_processor.py   | 18 ++---
 .../processor/test_base_state_processor.py    | 69 -------------------
 2 files changed, 9 insertions(+), 78 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 34723e4c109..717bd6dcd30 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
@@ -209,22 +209,22 @@ async def chain_updates(
         root_state: The root state of the app, no delta emitted if omitted.
     """
     from reflex.event import Event
-    from reflex.state import _resolve_delta
 
     ctx = EventContext.get()
 
     if root_state is not None:
-        # Snapshot and clean in one step, with no suspension between them: a
-        # concurrent write landing while the delta is being resolved or
-        # emitted must stay dirty for the next harvest, not be discarded by a
-        # clean that never snapshotted it. The delta is still emitted before
-        # the events below, so frontend events observe the latest state.
+        # Emit deltas first, so any frontend events are processed with the
+        # latest state. The clean deliberately runs after resolution: the
+        # SharedState fan-out captures its dirty vars at clean time, and
+        # resolving the delta is what re-marks linked vars through the patch
+        # machinery, so cleaning earlier would fan out a stale set (see
+        # tests/integration/test_linked_state.py).
         try:
-            delta = root_state.get_delta()
+            delta = await root_state._get_resolved_delta()
+            if delta:
+                await ctx.emit_delta(delta)
         finally:
             root_state._clean()
-        if delta and (delta := await _resolve_delta(delta)):
-            await ctx.emit_delta(delta)
 
     # Convert valid EventHandler and EventSpec into Event
     if fixed_events := Event.from_event_type(
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 a0561152ab3..cc0b1379e62 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
@@ -467,75 +467,6 @@ async def bg(self):
     )
 
 
-async def test_chain_updates_keeps_writes_made_during_delta_resolution(
-    wired_app: App,
-    real_base_state_processor: BaseStateEventProcessor,
-    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]],
-    token: str,
-):
-    """chain_updates must not clean a write it never snapshotted.
-
-    The snapshot and the clean happen in one step, before the resolved delta
-    is awaited or emitted, so a concurrent write landing during resolution
-    stays dirty for the next harvest even if a caller ever runs chain_updates
-    without holding the state lock.
-
-    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.
-    """
-    from reflex_base.event.processor.base_state_processor import chain_updates
-
-    resolving = asyncio.Event()
-    release = asyncio.Event()
-    hold_resolution = [False]
-
-    class MidResolveState(State):
-        victim: str = ""
-
-        @rx.var(cache=False)
-        async def window(self) -> int:
-            if hold_resolution[0]:
-                resolving.set()
-                await release.wait()
-            return 0
-
-    root_ctx = real_base_state_processor._root_context
-    assert root_ctx is not None
-    EventContext.set(root_ctx.fork(token=token))
-    state_manager = root_ctx.state_manager
-
-    try:
-        root = await state_manager.get_state(BaseStateToken(ident=token, cls=State))
-        substate = await root.get_state(MidResolveState)
-        root._clean()
-
-        hold_resolution[0] = True
-        flush = asyncio.ensure_future(
-            chain_updates(None, root_state=root, handler_name="unlocked_flush")
-        )
-        await resolving.wait()
-        substate.victim = "written"
-        hold_resolution[0] = False
-        release.set()
-        await flush
-
-        assert "victim" in substate.dirty_vars, (
-            "the write made during delta resolution was cleaned away"
-        )
-        await chain_updates(None, root_state=root, handler_name="second_flush")
-    finally:
-        State._always_dirty_substates.discard(MidResolveState.get_name())
-
-    state_name = MidResolveState.get_full_name()
-    victim_key = "victim" + FIELD_MARKER
-    assert any(
-        d.get(state_name, {}).get(victim_key) == "written" for _, d in emitted_deltas
-    ), f"the surviving write never reached a delta: {emitted_deltas}"
-
-
 async def test_chained_event_keeps_originating_router_data(
     wired_app: App,
     real_base_state_processor: BaseStateEventProcessor,

From 38d09d6e802ead502811c335a981bf2a033dc58e Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Fri, 21 Aug 2026 10:56:02 -0700
Subject: [PATCH 07/11] refactor(events): hoist the flush gate to module-level
 ensure_locked

Review feedback: the closure was redefined on every process_event call
just to capture two locals. A module-level function saves the
redefinition, is directly testable, and is reusable wherever a caller
needs to know whether a StateProxy currently holds the lock. Still
evaluated per yield, since a generator can move between inside and
outside its proxy context.
---
 .../event/processor/base_state_processor.py   | 64 +++++++++++--------
 .../processor/test_base_state_processor.py    | 32 ++++++++++
 2 files changed, 71 insertions(+), 25 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 717bd6dcd30..5fdd9e27094 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
@@ -233,6 +233,33 @@ async def chain_updates(
         await _route_events(ctx, fixed_events)
 
 
+def ensure_locked(
+    state: BaseState | StateProxy, root_state: BaseState | None
+) -> BaseState | None:
+    """The root to flush deltas from, only while the state lock is held.
+
+    Foreground handlers pass the locked root in. A background handler
+    yielding from inside ``async with self`` is suspended while its proxy
+    still holds the lock, so flushing through the proxy's root keeps the
+    documented ordering: deltas reach the frontend before the yielded event.
+    Outside the proxy context there is no lock, and flushing there is the
+    unlocked snapshot/clean that discards concurrent writes. Evaluated per
+    yield: a generator can move between inside and outside the context.
+
+    Args:
+        state: The state the handler runs against, possibly a StateProxy.
+        root_state: The locked root passed by foreground callers, if any.
+
+    Returns:
+        The root state to flush, or None when the lock is not held.
+    """
+    if root_state is not None:
+        return root_state
+    if isinstance(state, StateProxy) and state._is_mutable():
+        return state.__wrapped__._get_root_state()
+    return None
+
+
 async def process_event(
     handler: EventHandler,
     payload: dict,
@@ -269,25 +296,6 @@ async def process_event(
             f"Error transforming event payload for handler {handler_name}: {ex}"
         )
 
-    def _flush_root_state() -> BaseState | None:
-        """The root to flush deltas from, only while the state lock is held.
-
-        Foreground handlers pass the locked root in. A background handler
-        yielding from inside ``async with self`` is suspended while its proxy
-        still holds the lock, so flushing through the proxy's root keeps the
-        documented ordering: deltas reach the frontend before the yielded
-        event. Outside the proxy context there is no lock, and flushing there
-        is the unlocked snapshot/clean that discards concurrent writes.
-
-        Returns:
-            The root state to flush, or None when the lock is not held.
-        """
-        if root_state is not None:
-            return root_state
-        if isinstance(state, StateProxy) and state._is_mutable():
-            return state.__wrapped__._get_root_state()
-        return None
-
     # Handle async functions.
     if inspect.iscoroutinefunction(fn.func):
         events = await fn(**payload)
@@ -299,10 +307,12 @@ def _flush_root_state() -> BaseState | None:
     if inspect.isasyncgen(events):
         async for event in events:
             await chain_updates(
-                event, root_state=_flush_root_state(), handler_name=handler_name
+                event,
+                root_state=ensure_locked(state, root_state),
+                handler_name=handler_name,
             )
         await chain_updates(
-            None, root_state=_flush_root_state(), handler_name=handler_name
+            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
         )
 
     # Handle regular generators.
@@ -311,7 +321,7 @@ def _flush_root_state() -> BaseState | None:
             while True:
                 await chain_updates(
                     next(events),
-                    root_state=_flush_root_state(),
+                    root_state=ensure_locked(state, root_state),
                     handler_name=handler_name,
                 )
         except StopIteration as si:
@@ -319,16 +329,20 @@ def _flush_root_state() -> BaseState | None:
             # in the loop, we must catch StopIteration to access it
             if si.value is not None:
                 await chain_updates(
-                    si.value, root_state=_flush_root_state(), handler_name=handler_name
+                    si.value,
+                    root_state=ensure_locked(state, root_state),
+                    handler_name=handler_name,
                 )
         await chain_updates(
-            None, root_state=_flush_root_state(), handler_name=handler_name
+            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
         )
 
     # Handle regular event chains.
     else:
         await chain_updates(
-            events, root_state=_flush_root_state(), handler_name=handler_name
+            events,
+            root_state=ensure_locked(state, root_state),
+            handler_name=handler_name,
         )
 
 
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 cc0b1379e62..c0fa3ce82cc 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
@@ -536,3 +536,35 @@ def client_event(spec, router_data: dict[str, Any]) -> Event:
         BaseStateToken(ident=token, cls=State)
     )
     assert (await state.get_state(RouterState)).seen == ["/item/abc|abc"]
+
+
+async def test_ensure_locked_returns_a_root_only_while_the_lock_is_held(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """ensure_locked passes a locked root through and refuses everything else.
+
+    Foreground callers hand in the root they locked; a plain substate or an
+    un-entered proxy holds no lock, so there is nothing safe to flush.
+
+    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.event.processor.base_state_processor import ensure_locked
+
+    from reflex.istate.proxy import StateProxy
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    EventContext.set(root_ctx.fork(token=token))
+    root = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    substate = await root.get_state(OnLoadInternalState)
+
+    assert ensure_locked(substate, root) is root
+    assert ensure_locked(substate, None) is None
+    assert ensure_locked(StateProxy(substate), None) is None

From 202ad21c25b2f3e3c62ad64f61fee0aa86e687eb Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Fri, 21 Aug 2026 11:12:42 -0700
Subject: [PATCH 08/11] refactor(events): track context entry on StateProxy
 itself

Review feedback: the enter-tracking subclass existed only to record
whether the handler ever entered async-with-self, which is a fact about
the proxy. Track it natively on StateProxy and drop the subclass. The
flag is set at the top of __aenter__ so the parent-delegation branch
records entry too.
---
 .../event/processor/base_state_processor.py   | 25 +------------------
 reflex/istate/proxy.py                        |  4 +++
 2 files changed, 5 insertions(+), 24 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 5fdd9e27094..aec3de4a1d3 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
@@ -346,29 +346,6 @@ async def process_event(
         )
 
 
-class _EnterTrackingStateProxy(StateProxy):
-    """A StateProxy recording whether the handler ever entered ``async with self``."""
-
-    def __init__(self, *args, **kwargs):
-        """Create the proxy with the entered flag unset.
-
-        Args:
-            *args: Positional arguments for StateProxy.
-            **kwargs: Keyword arguments for StateProxy.
-        """
-        super().__init__(*args, **kwargs)
-        self._self_entered_context = False
-
-    async def __aenter__(self):
-        """Enter the mutability context, recording that it was entered.
-
-        Returns:
-            This proxy in mutable mode.
-        """
-        self._self_entered_context = True
-        return await super().__aenter__()
-
-
 class BaseStateEventProcessor(EventProcessor):
     """Event processor for BaseState-derived states.
 
@@ -479,7 +456,7 @@ async def _execute_event(
         # its _clean() would be discarded before any delta carries it. A
         # background task's own state changes are emitted (and cleaned) by its
         # `async with self` context exits, which re-acquire the lock.
-        proxy = _EnterTrackingStateProxy(substate)
+        proxy = StateProxy(substate)
         await process_event(
             handler=registered_handler.handler,
             state=proxy,
diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py
index ce2eef96834..fd1b498b87f 100644
--- a/reflex/istate/proxy.py
+++ b/reflex/istate/proxy.py
@@ -96,6 +96,9 @@ def __init__(
         self._self_actx_lock = asyncio.Lock()
         self._self_actx_lock_holder = None
         self._self_parent_state_proxy = parent_state_proxy
+        # Whether `async with self` was ever entered; a background handler that
+        # never did emitted no delta, so the processor flushes once for it.
+        self._self_entered_context = False
 
     def _is_mutable(self) -> bool:
         """Check if the state is mutable.
@@ -122,6 +125,7 @@ async def __aenter__(self) -> Self:
         Raises:
             ImmutableStateError: If the state is already mutable.
         """
+        self._self_entered_context = True
         if self._self_parent_state_proxy is not None:
             from reflex.state import State
 

From 704c53ede2344bb562f5eac85085e097ad0b7e36 Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Fri, 21 Aug 2026 11:14:49 -0700
Subject: [PATCH 09/11] chore: add root-package news fragment for the
 StateProxy change

---
 news/6920.bugfix.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 news/6920.bugfix.md

diff --git a/news/6920.bugfix.md b/news/6920.bugfix.md
new file mode 100644
index 00000000000..e344c10aa50
--- /dev/null
+++ b/news/6920.bugfix.md
@@ -0,0 +1 @@
+StateProxy records whether `async with self` was ever entered, so the event processor can preserve the delta flush for background handlers that never enter the context while no longer flushing a shared state tree without the token lock (which silently discarded concurrent handlers' writes).

From 93acf21abde58dcc336ba6fa0b73ae8a5b2fb6c1 Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Fri, 21 Aug 2026 11:23:26 -0700
Subject: [PATCH 10/11] fix(events): mark the proxy entered only after a
 successful context enter

Review feedback: setting the flag at the top of __aenter__ meant a
failed enter, swallowed by the handler, would skip the locked fallback
flush and strand preamble dirty vars like router_data. Entered now
means a context actually opened, whose exit will flush; both branches
set it after their enter succeeds, and the failure path is pinned by a
test.
---
 reflex/istate/proxy.py                        |  3 +-
 .../processor/test_base_state_processor.py    | 47 +++++++++++++++++++
 2 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/reflex/istate/proxy.py b/reflex/istate/proxy.py
index fd1b498b87f..919827949a2 100644
--- a/reflex/istate/proxy.py
+++ b/reflex/istate/proxy.py
@@ -125,7 +125,6 @@ async def __aenter__(self) -> Self:
         Raises:
             ImmutableStateError: If the state is already mutable.
         """
-        self._self_entered_context = True
         if self._self_parent_state_proxy is not None:
             from reflex.state import State
 
@@ -138,6 +137,7 @@ async def __aenter__(self) -> Self:
                     State.get_class_substate(self._self_substate_path)
                 ),
             )
+            self._self_entered_context = True
             return self
         current_task = asyncio.current_task()
         if (
@@ -157,6 +157,7 @@ async def __aenter__(self) -> Self:
             )
             mutable_state = await self._self_actx.__aenter__()
             self._self_mutable = True
+            self._self_entered_context = True
             super().__setattr__(
                 "__wrapped__", mutable_state.get_substate(self._self_substate_path)
             )
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 c0fa3ce82cc..c8d0476edb3 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
@@ -568,3 +568,50 @@ async def test_ensure_locked_returns_a_root_only_while_the_lock_is_held(
     assert ensure_locked(substate, root) is root
     assert ensure_locked(substate, None) is None
     assert ensure_locked(StateProxy(substate), None) is None
+
+
+async def test_failed_context_enter_does_not_mark_the_proxy_entered(
+    wired_app: App,
+    real_base_state_processor: BaseStateEventProcessor,
+    token: str,
+):
+    """A proxy whose enter failed still gets the compatibility flush.
+
+    The entered flag means "a context opened whose exit will flush". If
+    ``__aenter__`` raises before that and the handler swallows it, the
+    processor must still run the locked fallback flush, or preamble dirty
+    vars like router_data would never reach a delta.
+
+    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.istate.proxy import StateProxy
+
+    root_ctx = real_base_state_processor._root_context
+    assert root_ctx is not None
+    EventContext.set(root_ctx.fork(token=token))
+    root = await root_ctx.state_manager.get_state(
+        BaseStateToken(ident=token, cls=State)
+    )
+    substate = await root.get_state(OnLoadInternalState)
+
+    proxy = StateProxy(substate)
+
+    def raise_on_modify(*args, **kwargs):
+        msg = "state manager unavailable"
+        raise RuntimeError(msg)
+
+    original = root_ctx.state_manager.modify_state_with_links
+    object.__setattr__(
+        root_ctx.state_manager, "modify_state_with_links", raise_on_modify
+    )
+    try:
+        with pytest.raises(RuntimeError, match="state manager unavailable"):
+            async with proxy:
+                pass
+    finally:
+        object.__setattr__(root_ctx.state_manager, "modify_state_with_links", original)
+
+    assert proxy._self_entered_context is False

From 3d4975bd2a9ddbaf53c85a1e867e4216a72ddeea Mon Sep 17 00:00:00 2001
From: Khaleel Al-Adhami <khaleel.aladhami@gmail.com>
Date: Fri, 21 Aug 2026 11:24:13 -0700
Subject: [PATCH 11/11] docs: state the user-facing symptom in the root news
 fragment

---
 news/6920.bugfix.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/news/6920.bugfix.md b/news/6920.bugfix.md
index e344c10aa50..03088174338 100644
--- a/news/6920.bugfix.md
+++ b/news/6920.bugfix.md
@@ -1 +1 @@
-StateProxy records whether `async with self` was ever entered, so the event processor can preserve the delta flush for background handlers that never enter the context while no longer flushing a shared state tree without the token lock (which silently discarded concurrent handlers' writes).
+Fixed a race where a finishing background task could silently discard state updates made by a concurrently running event handler before they reached the frontend, leaving the UI stale until the next write. Background handlers that never enter `async with self` still emit their delta, now computed under the state lock.