diff --git a/.claude/references/architecture.md b/.claude/references/architecture.md
index 265432d77..6550db755 100644
--- a/.claude/references/architecture.md
+++ b/.claude/references/architecture.md
@@ -15,8 +15,10 @@ The reactive system is based on a **push-pull** model with three core abstractio
   sources (Value) it reads from.
 - **Dependents**: Each reactive source maintains a list of downstream consumers
   that depend on it. When a source invalidates, it notifies all dependents.
-- **ReactiveEnvironment**: Global singleton managing the reactive graph,
-  execution queue, and flush cycles.
+- **ReactiveEnvironment**: Global singleton managing the reactive graph, the
+  reactive effect queue, and rounds. Its class docstring in
+  `shiny/reactive/_core.py` defines the terms used for the reactive system:
+  reactive effect queue, round, idle, cycle, and output flush.
 
 Key implementation details:
 
@@ -25,8 +27,9 @@ Key implementation details:
 - `Effect_()` is a side-effect that re-executes when dependencies change
 - `event()` decorator suppresses reactive dependencies for specific reads
 - The reactive graph is built automatically through the Context's dependency tracking
-- Execution uses a priority queue to ensure correct invalidation ordering
-- In tests, `reactive.flush()` forces a synchronous flush of the reactive graph
+- Each round takes effects from the reactive effect queue in priority order
+- In tests, `await reactive.flush()` runs rounds until the reactive environment is
+  idle
 
 ## Session Hierarchy
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d1678a4c9..843055959 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Breaking changes
+
+* Reactive effects now run concurrently, following Shiny for R's model. A reactive flush starts each invalidated effect and no longer waits for an `async` effect's awaited part, so a slow `async` effect, calc, or render function no longer delays other sessions. Within a session, input changes and `reactive.invalidate_later()` still wait until all of that session's effects have finished, and outputs are sent together once they have. What app authors may notice:
+
+    * `async` effects interleave at each `await` instead of running one after another. `priority` orders when effects start, not when they finish. Runs of the same effect still don't overlap: a re-run waits for the previous run.
+
+    * `reactive.lock()` no longer pauses reactive processing: Shiny itself no longer takes it. To change reactive state from another `asyncio` task, set the value directly (a flush is scheduled automatically) and `await reactive.flush()` to wait until the resulting reactive work has finished.
+
+    * `await reactive.flush()` returns right away, without waiting for dependents, when called from within an effect or from a task started by an effect that is still running, since waiting there could deadlock. A task that outlives the effect that started it, such as an extended task's body, still waits.
+
+    * Message handlers (`session.set_message_handler()`) run in their own task, so a slow handler no longer holds up other messages from the client.
+
+    * Each session sends its outputs in a task of its own, once all of its effects have finished. A slow client, or a slow `session.on_flush()` callback, delays only its own session's output, not other sessions' or the next reactive flush.
+
+    * Setting a `reactive.value` schedules a flush, including when `set()` is called from another thread. The reactive graph itself still isn't thread-safe, so from another thread use `loop.call_soon_threadsafe(value.set, new_value)`.
+
+  (#2508)
+
 ### New features
 
 * `ui.sidebar()` gains a `role` parameter (`"form"`, `"search"`, `"complementary"`, or `"region"`) for opt-in ARIA landmark markup (rstudio/bslib#1359). `"complementary"` renders an `<aside>`, other roles render a `<div>` with the corresponding `role` attribute, and landmark roles require an accessible name (from `title`, `aria_label`, or `aria_labelledby`). Note that the default (`role=None`) now renders a neutral `<div>` instead of an `<aside>`. In addition, `ui.page_sidebar()` now places the whole sidebar layout inside the page's `<main>` landmark. (#2526)
@@ -17,6 +35,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug fixes
 
+* Setting a reactive value from a download handler (`@render.download_button`) now updates the outputs and effects that read it right away, including while a streamed download is still sending. The session also keeps handling input during a streamed download. Before, they updated only after the next message from the client. (#1785)
+
 * `Progress.set(0)` now places the bar at `0` relative to `min` and `max`, instead of sending `0` unnormalized. With a negative `min`, such as `Progress(min=-10, max=10)`, a value of `0` showed an empty bar rather than a half-full one. (#2518)
 
 * `ui.input_date()`'s `datesdisabled` now works when `format` is not the default `yyyy-mm-dd`. The dates are now converted on the client the same way `min`/`max` are, instead of being parsed by bootstrap-datepicker with the display `format`. The `data-date-dates-disabled` attribute is replaced by `data-dates-disabled` (and omitted when `datesdisabled` is `None`), and `controller.InputDate.expect_datesdisabled()` checks the new attribute. Requires the updated vendored `shiny.js` (rstudio/shiny#4434). (#2523)
diff --git a/pytest.ini b/pytest.ini
index 0dba85c53..db5362bc2 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -5,6 +5,14 @@ asyncio_default_fixture_loop_scope=function
 testpaths=tests/pytest/
 ; Note: Browsers are set within `./Makefile`
 addopts = --strict-markers --durations=6 --durations-min=5.0 --numprocesses auto
+; No test may take more than 2 minutes, so a hang fails instead of running until
+; CI cancels the job. The `thread` method kills the hung xdist worker (which
+; xdist reports as a failure of that test); its own stack dump is lost in the
+; worker's IPC pipe, so faulthandler dumps every thread's stack to stderr first.
+; (Same approach as the Playwright targets in the Makefile.)
+timeout = 120
+timeout_method = thread
+faulthandler_timeout = 110
 ; Tests must not emit stray warnings; assert expected ones with `pytest.warns()`.
 ; Third-party deprecations we can't act on go in the ignore list below.
 filterwarnings =
@@ -24,4 +32,7 @@ filterwarnings =
     ; test at fault -- i.e. they make the suite flaky instead of strict.
     ignore::ResourceWarning
     ignore::pytest.PytestUnraisableExceptionWarning
+    ; When a timeout kills an xdist worker, syrupy warns that the killed worker's
+    ; tests are missing; as an error, that hides the summary naming the hung test.
+    ignore:.*selected test\(s\) missing from collected items:UserWarning
 ; verbosity_test_cases=2
diff --git a/shiny/.agents/skills/shiny-for-python/references/extended-tasks.md b/shiny/.agents/skills/shiny-for-python/references/extended-tasks.md
index 5cb8b957b..91ab5e7ed 100644
--- a/shiny/.agents/skills/shiny-for-python/references/extended-tasks.md
+++ b/shiny/.agents/skills/shiny-for-python/references/extended-tasks.md
@@ -3,15 +3,18 @@
 ## Overview
 
 A slow computation inside a `@reactive.calc`, `@reactive.effect`, or `@render.*`
-blocks the reactive flush - even an `async def` one - so the whole session (and,
-for shared resources, other sessions) freezes until it returns. Do NOT put a
-slow API call, model inference, or long loop directly in a calc/effect.
+keeps its session busy until it returns: that session applies no input changes
+and sends no outputs in the meantime. A sync one also blocks every session in the
+process, since it holds the event loop. (Other sessions keep running while an
+`async def` one awaits.) Do NOT put a slow API call, model inference, or long loop
+directly in a calc/effect.
 
 `@reactive.extended_task` runs an **async** function in a background asyncio
-task, off the reactive flush. Reactivity keeps processing while it runs; the
-result flows back into the reactive graph when it finishes.
+task, outside the session's reactive cycle. The session keeps processing inputs
+and outputs while it runs; the result flows back into the reactive graph when it
+finishes.
 
-Because it runs outside the reactive flush, the task **cannot read reactive
+Because it runs outside the reactive graph, the task **cannot read reactive
 sources** (`input.x()`, a `reactive.value`, a calc). Read those values *before*
 invoking and pass them in as arguments; reading one inside raises an error.
 
@@ -121,6 +124,9 @@ current one finishes.
 - Long synchronous work (a blocking library call) inside the async task still
   blocks the event loop - offload it with `asyncio.to_thread(...)` or an async
   client.
+- Calling `value.set(...)` from that worker thread -> the reactive graph isn't
+  thread-safe and it can raise. Return the result to the task instead, or hand the
+  call to the event loop: `loop.call_soon_threadsafe(value.set, new_value)`.
 - Calling `task.result()` outside a reactive context -> no dependency is tracked
   and it errors; read it inside a `@render.*`, calc, or effect.
 - Expecting a fresh `.invoke()` to interrupt the running task -> it queues
diff --git a/shiny/_app.py b/shiny/_app.py
index 00647ed66..c3964e5b0 100644
--- a/shiny/_app.py
+++ b/shiny/_app.py
@@ -51,6 +51,7 @@
 )
 from .html_dependencies import _page_deps
 from .http_staticfiles import FileResponse, StaticFiles
+from .reactive._core import _reactive_environment
 from .session._session import AppSession, Inputs, Outputs, Session, session_context
 from .types import MISSING, MISSING_TYPE
 from .ui._page import DEPS_PLACEHOLDER, PageHtmlDocument, page_html
@@ -240,7 +241,10 @@ def __init__(
 
         self._sessions: dict[str, AppSession] = {}
 
-        # self._sessions_needing_flush: dict[int, AppSession] = {}
+        # Sessions with outputs, errors, or input messages to send after the next
+        # round (R's `appsNeedingFlush`).
+        self._sessions_needing_output_flush: dict[str, AppSession] = {}
+        self._unregister_output_flush_hook: Optional[Callable[[], None]] = None
 
         self._registered_dependencies: dict[str, HTMLDependency] = {}
         self._dependency_handler = starlette.routing.Router()
@@ -469,13 +473,28 @@ async def _on_session_request_cb(self, request: Request) -> ASGIApp:
         return JSONResponse({"detail": "Not Found"}, status_code=404)
 
     # ==========================================================================
-    # Flush
+    # Output flush
     # ==========================================================================
-    def _request_flush(self, session: AppSession) -> None:
-        # TODO: Until we have reactive domains, because we can't yet keep track
-        # of which sessions need a flush.
-        pass
-        # self._sessions_needing_flush[session.id] = session
+    def _request_output_flush(self, session: AppSession) -> None:
+        self._sessions_needing_output_flush[session.id] = session
+        if self._unregister_output_flush_hook is None:
+            self._unregister_output_flush_hook = (
+                _reactive_environment.on_round_finished(self._start_output_flushes)
+            )
+        _reactive_environment.request_round()
+
+    async def _start_output_flushes(self) -> None:
+        """
+        After each round, start an output flush for each requesting session.
+
+        Each session flushes in its own task (see `AppSession._start_output_flush()`),
+        so a slow client, or a slow `on_flush` callback, delays only its own session:
+        other sessions' output and the next round don't wait for it.
+        """
+        sessions = list(self._sessions_needing_output_flush.values())
+        self._sessions_needing_output_flush.clear()
+        for session in sessions:
+            session._start_output_flush()
 
     # ==========================================================================
     # HTML Dependency stuff
diff --git a/shiny/otel/_core.py b/shiny/otel/_core.py
index 7e781bfad..9591a5b5c 100644
--- a/shiny/otel/_core.py
+++ b/shiny/otel/_core.py
@@ -283,8 +283,8 @@ def detached_otel_context() -> Generator[None, None, None]:
     from shiny.otel._core import detached_otel_context
 
     with detached_otel_context():
-        ctx.invalidate()
-        await flush()
+        with tracer.start_as_current_span("timer_tick"):  # a root span
+            ...
     ```
     """
     token = otel_context.attach(otel_context.Context())
diff --git a/shiny/reactive/_core.py b/shiny/reactive/_core.py
index 32135ca6e..90baee931 100644
--- a/shiny/reactive/_core.py
+++ b/shiny/reactive/_core.py
@@ -13,8 +13,10 @@
 
 import asyncio
 import contextlib
+import contextvars
 import time
 import traceback
+import types
 import typing
 import warnings
 from contextvars import ContextVar
@@ -23,8 +25,8 @@
 from .. import _utils
 from .._datastructures import PriorityQueueFIFO
 from .._docstring import add_example, no_example
+from .._typing_extensions import TypeGuard
 from ..otel._collect import OtelCollectLevel, _get_env_level
-from ..otel._core import detached_otel_context
 from ..otel._span_wrappers import shiny_otel_span
 from ..types import MISSING, MISSING_TYPE
 
@@ -76,12 +78,12 @@ def on_invalidate(self, func: Callable[[], None]) -> None:
             self._invalidate_callbacks.append(func)
 
     def add_pending_flush(self, priority: int) -> None:
-        """Tell the reactive environment that this context should be flushed the
-        next time flushReact() called."""
-        _reactive_environment.add_pending_flush(self, priority)
+        """Add this context to the reactive effect queue, to run in the next round."""
+        _reactive_environment.enqueue_effect(self, priority)
 
     def on_flush(self, func: Callable[[], Awaitable[None]]) -> None:
-        """Register a function to be called when this context is flushed."""
+        """Register a function to be called when a round takes this context from the
+        reactive effect queue."""
         self._flush_callbacks.append(func)
 
     async def execute_flush_callbacks(self) -> None:
@@ -122,17 +124,67 @@ def invalidate(self) -> None:
             dep_ctx.invalidate()
 
 
+@types.coroutine
+def _yield_to_loop() -> Generator[None, None, None]:
+    # What `asyncio.sleep(0)` does, without going through a patchable function.
+    yield
+
+
+# The task running the current round, effect, or session output flush. Tasks created
+# from it inherit this value, which lets `run_until_idle()` tell whether it was called
+# from within a run that is still going (directly or through a task it started).
+_enclosing_run: ContextVar[Optional["asyncio.Task[object]"]] = ContextVar(
+    "enclosing_run", default=None
+)
+
+
+def _is_alive(
+    loop: Optional[asyncio.AbstractEventLoop],
+) -> TypeGuard[asyncio.AbstractEventLoop]:
+    return loop is not None and not loop.is_closed() and loop.is_running()
+
+
 class ReactiveEnvironment:
-    """The reactive environment"""
+    """
+    The reactive environment.
+
+    Terms used throughout the reactive system:
+
+    * **Reactive effect queue**: the single, process-wide priority queue of reactive
+      effect contexts waiting to run. Invalidating an effect adds it to this queue.
+    * **Round**: one drain of the reactive effect queue, followed by the
+      round-finished callbacks. A round starts each effect in its own task and does
+      not wait for any effect's async part.
+    * **Idle**: the reactive effect queue is empty, and no round, effect, or session
+      output flush task is running. `reactive.flush()` runs rounds until idle.
+    * **Cycle** (per session): one action (such as an input change), the effects it
+      starts, and the output flush that ends it. One cycle can span several rounds,
+      and one round can advance the cycles of several sessions.
+    * **Output flush** (per session): sending a session's outputs to the client,
+      once that session's effects have finished.
+    """
 
     def __init__(self) -> None:
         self._current_context: ContextVar[Optional[Context]] = ContextVar(
             "current_context", default=None
         )
         self._next_id: int = 0
-        self._pending_flush_queue: PriorityQueueFIFO[Context] = PriorityQueueFIFO()
+        self._effect_queue: PriorityQueueFIFO[Context] = PriorityQueueFIFO()
         self._lock: Optional[asyncio.Lock] = None
-        self._flushed_callbacks = _utils.AsyncCallbacks()
+        self._round_finished_callbacks = _utils.AsyncCallbacks()
+        self._round_requested: bool = False
+        # The event loop Shiny runs on, for requests made from other threads.
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._round_running: bool = False
+        # Resolved by the next round to start (see `run_next_round()`).
+        self._next_round_waiters: list[asyncio.Future[None]] = []
+        # Set when a round is requested while one is running.
+        self._rerun_round: bool = False
+        # Strong references to fire-and-forget tasks (rounds, effect runs, and session
+        # output flushes); the event loop only keeps weak ones.
+        self._tasks: set[asyncio.Task[None]] = set()
+        # Tasks stranded on an event loop that stopped (see `_adopt_loop()`).
+        self._abandoned_tasks: set[asyncio.Task[None]] = set()
 
     @property
     def lock(self) -> asyncio.Lock:
@@ -170,32 +222,186 @@ def current_context(self) -> Context:
             raise RuntimeError("No current reactive context")
         return ctx
 
-    def on_flushed(
+    def on_round_finished(
         self, func: Callable[[], Awaitable[None]], once: bool = False
     ) -> Callable[[], None]:
-        return self._flushed_callbacks.register(func, once=once)
-
-    async def flush(self) -> None:
-        """Flush all pending operations"""
-        # Wrap entire flush cycle in reactive_update span (or no-op if not collecting)
-        async with shiny_otel_span(
-            "reactive_update",
-            infer_session_id=True,
-            required_level=OtelCollectLevel.REACTIVE_UPDATE,
-            collection_level=_get_env_level(),
-        ):
-            await self._flush_sequential()
-            await self._flushed_callbacks.invoke()
-
-    async def _flush_sequential(self) -> None:
-        # Sequential flush: instead of storing the tasks in a list and calling gather()
-        # on them later, just run each effect in sequence.
-        while not self._pending_flush_queue.empty():
-            ctx = self._pending_flush_queue.get()
-            await ctx.execute_flush_callbacks()
-
-    def add_pending_flush(self, ctx: Context, priority: int) -> None:
-        self._pending_flush_queue.put(priority, ctx)
+        """Register a function to be called at the end of every round."""
+        return self._round_finished_callbacks.register(func, once=once)
+
+    async def start_round(self) -> None:
+        """
+        Start one round: start every context in the reactive effect queue, then
+        invoke the round-finished callbacks. If a round is already running, return
+        right away; the running round starts another when it finishes.
+
+        This never waits for an effect's async part: each context runs in its own
+        task. Priority orders when effects start, not when they finish.
+        """
+        # Don't start a second round while one is running; the running one requests
+        # another when it finishes.
+        self._adopt_loop(asyncio.get_running_loop())
+        if self._round_running:
+            self._rerun_round = True
+            return
+        self._round_running = True
+        self._rerun_round = False
+        self._round_requested = False
+        self._loop = asyncio.get_running_loop()
+        waiters = self._next_round_waiters
+        self._next_round_waiters = []
+        token = _enclosing_run.set(asyncio.current_task())
+        try:
+            # Wrap the round in a reactive_update span (or no-op if not collecting)
+            async with shiny_otel_span(
+                "reactive_update",
+                infer_session_id=True,
+                required_level=OtelCollectLevel.REACTIVE_UPDATE,
+                collection_level=_get_env_level(),
+            ):
+                while not self._effect_queue.empty():
+                    ctx = self._effect_queue.get()
+                    self._spawn(self._run_context(ctx))
+                    # CPython runs ready callbacks FIFO, so the task's sync part runs
+                    # now, and anything it invalidates is queued before we take the
+                    # next ctx.
+                    await _yield_to_loop()
+                await self._round_finished_callbacks.invoke()
+        finally:
+            _enclosing_run.reset(token)
+            self._round_running = False
+            for waiter in waiters:
+                if not waiter.done():
+                    waiter.set_result(None)
+            if (
+                self._rerun_round
+                or self._next_round_waiters
+                or not self._effect_queue.empty()
+            ):
+                self._round_requested = False
+                self.request_round()
+
+    async def _run_context(self, ctx: Context) -> None:
+        _enclosing_run.set(asyncio.current_task())
+        await ctx.execute_flush_callbacks()
+
+    async def run_next_round(self) -> None:
+        """
+        Cause a round that starts after this call, and return once it has finished.
+
+        Runs that round itself when none is running. Otherwise, requests another
+        round from the running one, and waits for that round to finish.
+        """
+        self._adopt_loop(asyncio.get_running_loop())
+        if not self._round_running:
+            await self.start_round()
+            return
+        waiter: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+        self._next_round_waiters.append(waiter)
+        self._rerun_round = True
+        await waiter
+
+    async def run_until_idle(self) -> None:
+        """
+        Run rounds until idle: the reactive effect queue is empty and no round,
+        effect, or session output flush task is running.
+
+        Returns right away, without running anything, when called from within a
+        round, effect, or output flush that is still going, including from a task it
+        started (e.g. via `asyncio.gather()` or `asyncio.wait_for()`): that run may
+        be waiting on the caller, so waiting here could deadlock. A task started by an effect that has since finished
+        (e.g. an extended task's body) waits as usual.
+        """
+        owner = _enclosing_run.get()
+        if owner is not None and not owner.done():
+            return
+        current = asyncio.current_task()
+        while True:
+            await self.run_next_round()
+            running = {t for t in self._tasks if not t.done() and t is not current}
+            if not running and self._effect_queue.empty():
+                return
+            if running:
+                await asyncio.wait(running)
+
+    def request_round(self) -> None:
+        """
+        Schedule a single round on the next event-loop iteration. Requests made before
+        it runs are merged into it.
+
+        Safe to call from another thread: the request is handed to the event loop
+        Shiny runs on. (That makes only the *request* thread-safe. Reactive state
+        itself must be changed on the loop's thread, e.g. with
+        `loop.call_soon_threadsafe(value.set, x)`.)
+        """
+        try:
+            loop: Optional[asyncio.AbstractEventLoop] = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        home = self._loop
+        if _is_alive(home) and loop is not home:
+            # Another thread: retry on the loop's own thread, so only that thread
+            # touches `_round_requested` and schedules the round.
+            try:
+                home.call_soon_threadsafe(self.request_round)
+            except RuntimeError:
+                pass  # The loop closed in the meantime.
+            return
+        if loop is None:
+            # No loop at all (e.g. a reactive graph built synchronously); whoever
+            # runs the graph will call `start_round()` explicitly.
+            return
+        self._adopt_loop(loop)
+        self._loop = loop
+        if self._round_requested:
+            return
+        self._round_requested = True
+        # A fresh context keeps the requester's session and OTel span out of the round.
+        loop.call_soon(self._start_requested_round, context=contextvars.Context())
+
+    def _adopt_loop(self, loop: asyncio.AbstractEventLoop) -> None:
+        """
+        Discard round state left by an event loop that has stopped.
+
+        A round (or request) stranded on a loop that stopped mid-round, such as a
+        previous `test_server()` run's, can never finish. Without this, later rounds
+        would return as if one were running, and `run_next_round()` would wait
+        forever.
+        """
+        old = self._loop
+        if old is None or old is loop or _is_alive(old):
+            return
+        self._round_running = False
+        self._rerun_round = False
+        self._round_requested = False
+        self._next_round_waiters = []
+        # Stop waiting on the dead loop's tasks, but keep them: if collected, their
+        # coroutines' `finally` blocks would run (and fail) in whatever runs then.
+        # ponytail: never freed; only loops that stop mid-round (tests, repeated
+        # `test_server()` runs) leave any.
+        self._abandoned_tasks |= self._tasks
+        self._tasks = set()
+        self._loop = loop
+
+    def _start_requested_round(self) -> None:
+        if self._round_requested:
+            self._spawn(self.start_round())
+
+    def _spawn(self, coro: Awaitable[None]) -> "asyncio.Task[None]":
+        task = asyncio.ensure_future(coro)
+        self._tasks.add(task)
+        task.add_done_callback(self._on_task_done)
+        return task
+
+    def _on_task_done(self, task: asyncio.Task[None]) -> None:
+        self._tasks.discard(task)
+        if not task.cancelled() and (err := task.exception()) is not None:
+            # Effects report their own errors; this only catches bugs in a round or
+            # output flush.
+            traceback.print_exception(type(err), err, err.__traceback__)
+
+    def enqueue_effect(self, ctx: Context, priority: int) -> None:
+        self._effect_queue.put(priority, ctx)
+        self.request_round()
 
     @contextlib.contextmanager
     def isolate(self) -> Generator[None, None, None]:
@@ -266,8 +472,21 @@ async def flush() -> None:
     -------
     You shouldn't ever need to call this function inside of a Shiny app. It's only
     useful for testing and running reactive code interactively in the console.
+
+    Runs rounds until the reactive environment is idle: the reactive effect queue is
+    empty, every started effect (including its async part) has finished, and each
+    session's resulting outputs have been sent.
+
+    Note
+    ----
+    Called from within an effect, or from a task started by an effect that is still
+    running (e.g. via `asyncio.create_task()` or `asyncio.gather()`), this returns
+    right away without waiting: the effect may be waiting on the caller, so waiting
+    could deadlock. Dependents still run in the next round, but code right after
+    this call can't rely on them having run yet. A task that outlives the effect
+    that started it (such as an extended task's body) waits as usual.
     """
-    await _reactive_environment.flush()
+    await _reactive_environment.run_until_idle()
 
 
 @no_example()
@@ -275,12 +494,16 @@ def on_flushed(
     func: Callable[[], Awaitable[None]], once: bool = False
 ) -> Callable[[], None]:
     """
-    Register a function to be called when the reactive environment is flushed.
+    Register a function to be called at the end of every round.
+
+    A round starts every effect in the reactive effect queue, without waiting for
+    their async parts. One call to :func:`~shiny.reactive.flush` can run several
+    rounds.
 
     Parameters
     ----------
     func
-        The function to be called when the reactive environment is flushed
+        The function to be called at the end of every round.
     once
         Should the function be run once, and then cleared, or should it
         re-run each time the event occurs.
@@ -295,22 +518,30 @@ def on_flushed(
     * :func:`~shiny.reactive.flush`
     """
 
-    return _reactive_environment.on_flushed(func, once)
+    return _reactive_environment.on_round_finished(func, once)
 
 
 @no_example()
 def lock() -> asyncio.Lock:
     """
-    A lock that should be held whenever manipulating the reactive graph.
+    A process-wide lock, kept for backward compatibility.
+
+    Shiny no longer takes this lock itself, so holding it doesn't pause reactive
+    processing or serialize effects. Code that holds it only excludes other code
+    that also holds it.
 
-    For example, :func:`~shiny.reactive.lock` makes it safe to set a
-    :class:`~reactive.value` and call :func:`~shiny.reactive.flush` from a different
-    :class:`~asyncio.Task` than the one that is running the Shiny
-    :class:`~shiny.Session`.
+    To change reactive state from a different :class:`~asyncio.Task` than the one
+    running the Shiny :class:`~shiny.Session`, set the :class:`~reactive.value`
+    directly; a round is scheduled automatically. Await
+    :func:`~shiny.reactive.flush` to wait until the resulting reactive work has
+    finished.
     """
     return _reactive_environment.lock
 
 
+_timer_tasks: set[asyncio.Task[None]] = set()
+
+
 @add_example()
 def invalidate_later(
     delay: float, *, session: "MISSING_TYPE | Session | None" = MISSING
@@ -369,18 +600,18 @@ async def _task(ctx: Context, deadline: float) -> None:
                 # only be a no-op.
                 return
 
-            async with lock():
-                # Prevent the ctx.invalidate() from killing our own task. (Another way
-                # to accomplish this is to unregister our ctx.on_invalidate handler, but
-                # ctx.on_invalidate doesn't currently allow unregistration.)
-                cancellable = False
+            # Prevent the ctx.invalidate() from killing our own task. (Another way
+            # to accomplish this is to unregister our ctx.on_invalidate handler, but
+            # ctx.on_invalidate doesn't currently allow unregistration.)
+            cancellable = False
 
-                # Detach from any active OTel span so the flush's reactive_update
-                # span starts as a root span. The flush is timer-driven, not
-                # caused by a user action, so it should have no parent.
-                with detached_otel_context():
-                    ctx.invalidate()
-                    await flush()
+            # Like an input change, the invalidation waits until the session is idle.
+            # The resulting round is requested with a fresh context, so its
+            # reactive_update span has no parent.
+            if session:
+                session._cycle_start_action(ctx.invalidate)
+            else:
+                ctx.invalidate()
 
         except BaseException:
             traceback.print_exc()
@@ -390,6 +621,10 @@ async def _task(ctx: Context, deadline: float) -> None:
                 unsub()
 
     task = asyncio.create_task(_task(ctx, deadline))
+    # Keep a strong reference; not in `_reactive_environment._tasks`, since a timer
+    # that re-arms itself would keep `run_until_idle()` running forever.
+    _timer_tasks.add(task)
+    task.add_done_callback(_timer_tasks.discard)
 
     def cancel_task():
         if cancellable and not task.cancelled():
diff --git a/shiny/reactive/_extended_task.py b/shiny/reactive/_extended_task.py
index 2227c6db0..0aef13173 100644
--- a/shiny/reactive/_extended_task.py
+++ b/shiny/reactive/_extended_task.py
@@ -24,7 +24,7 @@
 from ..otel._function_attrs import resolve_func_otel_level
 from ..otel._labels import create_otel_span_name
 from ..otel._span_wrappers import shiny_otel_span
-from ._core import Context, flush, lock
+from ._core import Context, _reactive_environment
 from ._reactives import Value, isolate
 
 P = ParamSpec("P")
@@ -43,7 +43,8 @@
 class DenialContext(Context):
     """
     A context that denies all requests to read reactive sources. We use this to ensure
-    that code run outside of reactive.lock doesn't inadvertedly read reactive sources.
+    that an extended task's body, which runs outside any round, doesn't
+    inadvertently read reactive sources.
     """
 
     def on_invalidate(self, func: Callable[[], None]) -> None:
@@ -190,7 +191,7 @@ def _done_callback(self, task: asyncio.Task[R]) -> None:
         """
         Several things must happen when a task finishes:
         1. Update the status and value/error.
-        2. Reactive flush, so that reactive dependencies can move forward.
+        2. Wait for the next round, so that reactive dependencies can move forward.
         3. If there are queued up invocations, run the next one.
         """
 
@@ -202,22 +203,24 @@ def _done_callback(self, task: asyncio.Task[R]) -> None:
         self._task = None
 
         async def _impl():
-            async with lock():
-                if task.cancelled():
-                    self.status.set("cancelled")
-                elif task.exception() is not None:
-                    self.error.set(cast(BaseException, task.exception()))
-                    self.status.set("error")
-                else:
-                    self.value.set(task.result())
-                    self.status.set("success")
-
-                await flush()
-
-                if len(self._invocation_queue) > 0:
-                    next_invocation = self._invocation_queue.pop(0)
-                    next_invocation()
-                    await flush()
+            if task.cancelled():
+                self.status.set("cancelled")
+            elif task.exception() is not None:
+                self.error.set(cast(BaseException, task.exception()))
+                self.status.set("error")
+            else:
+                self.value.set(task.result())
+                self.status.set("success")
+
+            # Start the dependents (without waiting for their async parts) so they
+            # see this result before the next invocation replaces it. If a round is
+            # already running, it may have passed them already, so this requests
+            # another round and waits for it.
+            await _reactive_environment.run_next_round()
+
+            if len(self._invocation_queue) > 0:
+                next_invocation = self._invocation_queue.pop(0)
+                next_invocation()
 
         done_task = asyncio.create_task(_impl())
         self._done_tasks.add(done_task)
diff --git a/shiny/reactive/_reactives.py b/shiny/reactive/_reactives.py
index c3e0304d0..6cc49d51b 100644
--- a/shiny/reactive/_reactives.py
+++ b/shiny/reactive/_reactives.py
@@ -58,6 +58,14 @@
 from ._utils import is_user_code_frame
 
 
+def _current_task() -> Optional[asyncio.Task[object]]:
+    try:
+        return asyncio.current_task()
+    except RuntimeError:
+        # No running loop (a sync calc read outside of asyncio).
+        return None
+
+
 def _weak_destroy_callback(
     method: Callable[[], None],
     session: Session,
@@ -493,6 +501,16 @@ def set(self, value: T) -> bool:
             If the value has been destroyed.
         RuntimeError
             If called on a read-only reactive value.
+
+        Note
+        ----
+        Setting a new value schedules a reactive flush, so its dependents re-run
+        without calling :func:`~shiny.reactive.flush`. Call ``set()`` from the thread
+        running Shiny's event loop. From any other thread, hand the call to that loop
+        with ``loop.call_soon_threadsafe(value.set, new_value)``. The reactive graph
+        isn't safe to change from two threads at once, and calling ``set()`` directly
+        from another thread can fail (e.g. when a session's effect depends on the
+        value).
         """
         if self._destroyed:
             raise DestroyedReactiveError(
@@ -671,6 +689,10 @@ def __init__(
         self._dependents: Dependents = Dependents()
         self._invalidated: bool = True
         self._running: bool = False
+        # The task running `update_value()`, and readers from other tasks waiting on
+        # that run (async calcs only: concurrent effects can read mid-run).
+        self._running_task: Optional[asyncio.Task[object]] = None
+        self._run_waiters: list[asyncio.Future[None]] = []
         self._most_recent_ctx_id: int = -1
         self._ctx: Optional[Context] = None
         self._exec_count: int = 0
@@ -770,8 +792,23 @@ async def get_value(self) -> T:
             )
         self._dependents.register()
 
+        # Another task is computing this value: share that run rather than starting a
+        # second one. Recompute only if it was invalidated meanwhile.
+        current = _current_task()
+        while (
+            self._running
+            and not self._invalidated
+            and self._running_task is not current
+        ):
+            waiter: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+            self._run_waiters.append(waiter)
+            await waiter
+
         if self._invalidated or self._running:
-            await self.update_value()
+            value, error = await self._update_value()
+            if error:
+                raise error[0]
+            return value[0]
 
         if self._error:
             raise self._error[0]
@@ -780,16 +817,28 @@ async def get_value(self) -> T:
 
     # TODO: should this be private?
     async def update_value(self) -> None:
-        self._ctx = Context()
-        self._most_recent_ctx_id = self._ctx.id
+        await self._update_value()
+
+    async def _update_value(self) -> tuple[list[T], list[Exception]]:
+        """
+        Run the calc; return this run's value or error.
 
-        self._ctx.on_invalidate(self._on_invalidate_cb)
+        Only the most recent run is cached: an async run can be invalidated and
+        superseded by a newer one while it awaits.
+        """
+        ctx = Context()
+        self._ctx = ctx
+        self._most_recent_ctx_id = ctx.id
+
+        ctx.on_invalidate(self._on_invalidate_cb)
 
         self._exec_count += 1
         self._invalidated = False
 
-        was_running = self._running
         self._running = True
+        self._running_task = _current_task()
+        value: list[T] = []
+        error: list[Exception] = []
 
         from ..session import session_context
 
@@ -802,10 +851,25 @@ async def update_value(self) -> None:
                 collection_level=self._otel_level,
             ):
                 try:
-                    with self._ctx():
-                        await self._run_func()
+                    with ctx():
+                        try:
+                            value.append(await self._fn())
+                        except Exception as err:
+                            error.append(err)
                 finally:
-                    self._running = was_running
+                    if ctx.id == self._most_recent_ctx_id:
+                        # If this run was invalidated meanwhile, `_invalidated` makes
+                        # the next read recompute anyway.
+                        self._value[:] = value
+                        self._error[:] = error
+                        self._running = False
+                        self._running_task = None
+                        waiters = self._run_waiters
+                        self._run_waiters = []
+                        for waiter in waiters:
+                            if not waiter.done():
+                                waiter.set_result(None)
+        return value, error
 
     def _on_invalidate_cb(self) -> None:
         self._invalidated = True
@@ -813,24 +877,6 @@ def _on_invalidate_cb(self) -> None:
         self._dependents.invalidate()
         self._ctx = None  # Allow context to be GC'd
 
-    async def _run_func(self) -> None:
-        self._error.clear()
-        try:
-            val = await self._fn()
-
-            # Clear before appending: a reentrant recursive call to this same
-            # `Calc_` (the calc calling itself) re-enters `_run_func` while an
-            # outer frame's call is still pending, so without clearing first,
-            # each level's value just piles onto the same list and the very
-            # first (deepest/base-case) entry -- not this call's own result --
-            # would be whatever a stale index read back. Clearing keeps
-            # exactly one entry: the one this call just computed, read
-            # immediately afterward with no other call able to interleave.
-            self._value.clear()
-            self._value.append(val)
-        except Exception as err:
-            self._error.append(err)
-
     def _extract_otel_attrs(self, fn: Callable[..., Any]) -> SourceRefAttrs:
         """Extract OpenTelemetry attributes from the reactive function."""
         return extract_source_ref(fn)
@@ -1004,6 +1050,8 @@ def __init__(
         self._priority: int = priority
         self._suspended = suspended
         self._on_resume: Callable[[], None] = lambda: None
+        # Resolved when the latest run finishes (see `on_flush_cb`).
+        self._run_done: Optional[asyncio.Future[None]] = None
 
         self._invalidate_callbacks: list[Callable[[], None]] = []
         self._destroyed: bool = False
@@ -1056,7 +1104,7 @@ def __init__(
         # Extract collection level from function attribute (e.g., set by `@otel.suppress` or `@otel.collect` decorators)
         self._otel_level: OtelCollectLevel = resolve_func_otel_level(fn)
 
-        # Defer the first running of this until flushReact is called
+        # Defer the first run to the next round
         self._create_context().invalidate()
 
     def _create_context(self) -> Context:
@@ -1088,10 +1136,26 @@ def _continue() -> None:
                 _continue()
 
         async def on_flush_cb() -> None:
-            if not self._destroyed:
-                await self._run()
-            if self._session:
-                self._session._decrement_busy_count()
+            # Unlike R, runs of one effect don't overlap: a re-run waits for the
+            # previous run's async part, so a slower, older run can't finish last.
+            previous = self._run_done
+            done: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+            self._run_done = done
+            try:
+                if previous is not None and not previous.done():
+                    await asyncio.shield(previous)
+                if not self._destroyed:
+                    await self._run()
+            except Exception as e:
+                # The effect runs in its own task, so report errors to its session.
+                if not self._session:
+                    raise
+                await self._session._unhandled_error(e)
+            finally:
+                done.set_result(None)
+                # Every exit path (raised, cancelled) must end the busy period.
+                if self._session:
+                    self._session._decrement_busy_count()
 
         ctx.on_invalidate(on_invalidate_cb)
         ctx.on_flush(on_flush_cb)
@@ -1180,10 +1244,10 @@ def suspend(self) -> None:
         """
         Suspend the effect.
 
-        Pauses scheduling of flushes (re-executions) in response to invalidations. If
-        the effect was invalidated prior to this call but it has not re-executed yet
-        (because it waits until on_flush is called) then that re-execution will still
-        occur, because the flush is already scheduled.
+        Stops adding the effect to the reactive effect queue in response to
+        invalidations. If the effect was invalidated prior to this call but it has not
+        re-executed yet (because it waits for the next round), then that re-execution
+        will still occur, because the effect is already in the queue.
         """
         self._suspended = True
 
@@ -1192,8 +1256,8 @@ def resume(self) -> None:
         Resume the effect.
 
         Causes this effect to start re-executing in response to invalidations. If the
-        effect was invalidated while suspended, then it will schedule itself for
-        re-execution (pending flush).
+        effect was invalidated while suspended, then it adds itself to the reactive
+        effect queue, to re-execute in the next round.
         """
         if self._suspended:
             self._suspended = False
@@ -1208,8 +1272,10 @@ def set_priority(self, priority: int = 0) -> None:
         ----------
         priority
             The new priority. A higher value means higher priority: an effect with a
-            higher priority value will execute before all effects with lower priority
-            values. Positive, negative, and zero values are allowed.
+            higher priority value starts before all effects with lower priority
+            values. An async effect's awaited part doesn't hold back lower-priority
+            effects, so they can finish first. Positive, negative, and zero values
+            are allowed.
 
         Note
         ----
@@ -1269,8 +1335,9 @@ def effect(
         until resumed and invalidated).
     priority
         The new priority. A higher value means higher priority: an effect with a higher
-        priority value will execute before all effects with lower priority values.
-        Positive, negative, and zero values are allowed.
+        priority value starts before all effects with lower priority values. An async
+        effect's awaited part doesn't hold back lower-priority effects, so they can
+        finish first. Positive, negative, and zero values are allowed.
     session
         A :class:`~shiny.Session` instance. If not provided, the session is inferred via
         :func:`~shiny.session.get_current_session`.
diff --git a/shiny/session/_session.py b/shiny/session/_session.py
index 1539d22fd..460ae299f 100644
--- a/shiny/session/_session.py
+++ b/shiny/session/_session.py
@@ -67,11 +67,8 @@
 from ..otel._function_attrs import resolve_func_otel_level
 from ..otel._labels import create_otel_label, create_otel_span_name
 from ..otel._span_wrappers import shiny_otel_span, shiny_otel_span_stream
-from ..reactive import Effect_, Value, effect
-from ..reactive import flush as reactive_flush
-from ..reactive import isolate
-from ..reactive._core import lock
-from ..reactive._core import on_flushed as reactive_on_flushed
+from ..reactive import Effect_, Value, effect, isolate
+from ..reactive._core import _enclosing_run, _reactive_environment
 from ..render.renderer import Renderer, RendererT
 from ..testmode import _snapshot_preprocess_file_input
 from ..types import (
@@ -189,6 +186,9 @@ def __init__(self, record_test_values: bool = False):
         computes a value or errors again.
         """
 
+    def is_empty(self) -> bool:
+        return not self.values and not self.errors and not self.input_messages
+
     def reset(self) -> None:
         self.values.clear()
         self.errors.clear()
@@ -624,7 +624,13 @@ def on_flush(
         once: bool = True,
     ) -> Callable[[], None]:
         """
-        Register a function to call before the next reactive flush.
+        Register a function to call before this session next sends its outputs.
+
+        A session sends its outputs (values, errors, and input messages) in one
+        message once each cycle ends, i.e. once all of its effects have finished. It
+        sends them in a task of its own, so a slow callback here (or a slow client)
+        delays only this session. Anything the callback queues, such as a
+        `send_input_message()`, goes out in that same message.
 
         Parameters
         ----------
@@ -647,7 +653,9 @@ def on_flushed(
         once: bool = True,
     ) -> Callable[[], None]:
         """
-        Register a function to call after the next reactive flush.
+        Register a function to call after this session next sends its outputs.
+
+        See :meth:`on_flush` for when a session sends its outputs.
 
         Parameters
         ----------
@@ -818,6 +826,10 @@ def _increment_busy_count(self) -> None: ...
     @abstractmethod
     def _decrement_busy_count(self) -> None: ...
 
+    def _cycle_start_action(self, action: Callable[[], None]) -> None:
+        """Run `action` once the session is idle. Sessions without cycles run it now."""
+        action()
+
 
 def _print_exception(e: Exception) -> None:
     traceback.print_exception(type(e), e, e.__traceback__)
@@ -913,6 +925,21 @@ def __init__(
         self._conn: Connection = conn
         self._debug: bool = debug
         self._busy_count: int = 0
+        # Actions (input updates, timer invalidations) waiting for the session to be
+        # idle.
+        self._cycle_start_action_queue: list[Callable[[], None]] = []
+        # Whether a cycle has finished since outputs were last sent. Every cycle ends
+        # with a message, even an empty one. True so the first output flush always
+        # sends.
+        self._cycle_ended: bool = True
+        # Whether output flush requests are honored: from init until the session loop
+        # exits.
+        self._output_flush_enabled: bool = False
+        self._dispatch_tasks: set[asyncio.Task[None]] = set()
+        self._running_flush_callbacks: bool = False
+        # The task sending this session's outputs, and whether to run it once more.
+        self._output_flush_task: Optional[asyncio.Task[None]] = None
+        self._output_flush_again: bool = False
         self._message_handlers: dict[
             str,
             tuple[Callable[..., Awaitable[Jsonifiable]], Session],
@@ -1076,86 +1103,86 @@ def verify_state(expected_state: ConnectionState) -> None:
                         )
                         return
 
-                    async with lock():
-                        if message_obj["method"] == "init":
-                            verify_state(ConnectionState.Start)
+                    if message_obj["method"] == "init":
+                        verify_state(ConnectionState.Start)
 
-                            # BOOKMARKS!
-                            if not isinstance(self.bookmark, BookmarkApp):
-                                raise RuntimeError("`.bookmark` must be a BookmarkApp")
+                        # BOOKMARKS!
+                        if not isinstance(self.bookmark, BookmarkApp):
+                            raise RuntimeError("`.bookmark` must be a BookmarkApp")
 
-                            if ".clientdata_url_search" in message_obj["data"]:
-                                self.bookmark._set_restore_context(
-                                    await RestoreContext.from_query_string(
-                                        message_obj["data"][".clientdata_url_search"],
-                                        app=self.app,
-                                    )
+                        if ".clientdata_url_search" in message_obj["data"]:
+                            self.bookmark._set_restore_context(
+                                await RestoreContext.from_query_string(
+                                    message_obj["data"][".clientdata_url_search"],
+                                    app=self.app,
                                 )
-                            else:
-                                self.bookmark._set_restore_context(RestoreContext())
+                            )
+                        else:
+                            self.bookmark._set_restore_context(RestoreContext())
 
-                            # When a reactive flush occurs, flush the session's outputs,
-                            # errors, etc. to the client. Note that this is
-                            # `reactive._core.on_flushed`, not `self.on_flushed`.
-                            unreg = reactive_on_flushed(self._flush)
-                            # When the session ends, stop flushing outputs on reactive
-                            # flush.
-                            stack.callback(unreg)
+                        # From now on, output flush requests send this session's
+                        # outputs, errors, etc. to the client after the next round.
+                        # When the session ends, stop.
+                        self._output_flush_enabled = True
+                        stack.callback(self._disable_output_flush)
 
-                            # Set up bookmark callbacks here
-                            self.bookmark._create_effects()
+                        # Set up bookmark callbacks here
+                        self.bookmark._create_effects()
 
-                            conn_state = ConnectionState.Running
-                            message_obj = typing.cast(ClientMessageInit, message_obj)
-                            self._manage_inputs(message_obj["data"])
+                        conn_state = ConnectionState.Running
+                        message_obj = typing.cast(ClientMessageInit, message_obj)
+                        self._manage_inputs(message_obj["data"])
 
-                            # Wrap server function initialization in session_start span
-                            with session_context(self):
-                                async with shiny_otel_span(
-                                    "session_start",
-                                    attributes=lambda: extract_http_attributes(
-                                        self.http_conn
-                                    ),
-                                    required_level=OtelCollectLevel.SESSION,
-                                    collection_level=_get_env_level(),
-                                    infer_session_id=True,
-                                ):
-                                    self.app.server(self.input, self.output, self)
-
-                                    # Flush here to attempt a reactive_update within `session_start` otel span.
-                                    # Might also fix https://github.com/posit-dev/py-shiny/issues/1889
-                                    await reactive_flush()
-
-                        elif message_obj["method"] == "update":
-                            verify_state(ConnectionState.Running)
-
-                            message_obj = typing.cast(ClientMessageUpdate, message_obj)
+                        # Wrap server function initialization in session_start span
+                        with session_context(self):
+                            async with shiny_otel_span(
+                                "session_start",
+                                attributes=lambda: extract_http_attributes(
+                                    self.http_conn
+                                ),
+                                required_level=OtelCollectLevel.SESSION,
+                                collection_level=_get_env_level(),
+                                infer_session_id=True,
+                            ):
+                                self.app.server(self.input, self.output, self)
+
+                                # Start the first cycle within the `session_start`
+                                # span. This doesn't wait for async effects.
+                                self._request_output_flush()
+                                await _reactive_environment.start_round()
+
+                    elif message_obj["method"] == "update":
+                        verify_state(ConnectionState.Running)
+
+                        data = typing.cast(ClientMessageUpdate, message_obj)["data"]
+
+                        def manage_inputs(data: dict[str, object] = data) -> None:
                             # Set the session context for otel logging purposes
                             with session_context(self):
-                                self._manage_inputs(message_obj["data"])
+                                self._manage_inputs(data)
 
-                        elif "tag" in message_obj and "args" in message_obj:
-                            verify_state(ConnectionState.Running)
+                        # Inputs change only between cycles, so effects that are
+                        # still running keep seeing stable input values.
+                        self._cycle_start_action(manage_inputs)
 
-                            message_obj = typing.cast(ClientMessageOther, message_obj)
-                            await self._dispatch(message_obj)
+                    elif "tag" in message_obj and "args" in message_obj:
+                        verify_state(ConnectionState.Running)
 
-                        else:
-                            raise ProtocolError(
-                                f"Unrecognized method {message_obj['method']}"
-                            )
+                        # Handlers can be slow; run them in their own task so the
+                        # receive loop stays responsive. Responses carry the tag.
+                        # `reactive.flush()` doesn't wait on these tasks.
+                        task = asyncio.create_task(
+                            self._dispatch(typing.cast(ClientMessageOther, message_obj))
+                        )
+                        self._dispatch_tasks.add(task)
+                        task.add_done_callback(self._dispatch_tasks.discard)
 
-                        # Progress messages (of the "{binding: {id: xxx}}"" variety) may
-                        # have queued up at this point; let them drain before we send
-                        # the next message.
-                        # https://github.com/posit-dev/py-shiny/issues/1381
-                        await asyncio.sleep(0)
+                    else:
+                        raise ProtocolError(
+                            f"Unrecognized method {message_obj['method']}"
+                        )
 
-                        # Keep session context active for reactive flush so
-                        # reactive_update spans can consistently include session.id.
-                        with session_context(self):
-                            self._request_flush()
-                            await reactive_flush()
+                    self._request_output_flush()
 
             except ConnectionClosed:
                 ...
@@ -1308,11 +1335,19 @@ async def uploadEnd(job_id: str, input_id: str) -> None:
     async def _handle_request(
         self, request: Request, action: str, subpath: Optional[str]
     ) -> ASGIApp:
-        self._increment_busy_count()
+        # Busy pauses this session's cycle (no input updates, timers, or outputs), so
+        # only downloads take it, as in R; not uploads or dynamic routes. It covers
+        # only the call to the download handler: a streamed body is read after this
+        # returns, so values the stream sets reach the client while it streams, and
+        # the session keeps handling input during a long download.
+        busy = action == "download"
+        if busy:
+            self._increment_busy_count()
         try:
             return await self._handle_request_impl(request, action, subpath)
         finally:
-            self._decrement_busy_count()
+            if busy:
+                self._decrement_busy_count()
 
     async def _handle_request_impl(
         self, request: Request, action: str, subpath: Optional[str]
@@ -1572,7 +1607,7 @@ async def _build_test_snapshot(
 
     def send_input_message(self, id: str, message: dict[str, object]) -> None:
         self._outbound_message_queues.add_input_message(id, message)
-        self._request_flush()
+        self._request_output_flush()
 
     def _send_insert_ui(
         self, selector: str, multiple: bool, where: str, content: RenderedDeps
@@ -1634,7 +1669,9 @@ def on_flush(
         fn: Callable[[], None] | Callable[[], Awaitable[None]],
         once: bool = True,
     ) -> Callable[[], None]:
-        return self._flush_callbacks.register(wrap_async(fn), once)
+        unregister = self._flush_callbacks.register(wrap_async(fn), once)
+        self._request_output_flush()
+        return unregister
 
     def on_flushed(
         self,
@@ -1643,34 +1680,98 @@ def on_flushed(
     ) -> Callable[[], None]:
         return self._flushed_callbacks.register(wrap_async(fn), once)
 
-    def _request_flush(self) -> None:
-        self.app._request_flush(self)
+    def _request_output_flush(self) -> None:
+        """Send this session's outputs after the next round."""
+        # Anything queued by this session's own `on_flush` callbacks goes out in the
+        # message being built, so it needs no further output flush (which could loop).
+        if self._output_flush_enabled and not self._running_flush_callbacks:
+            self.app._request_output_flush(self)
+
+    def _disable_output_flush(self) -> None:
+        self._output_flush_enabled = False
+        self.app._sessions_needing_output_flush.pop(self.id, None)
+
+    def _start_output_flush(self) -> None:
+        """
+        Send this session's outputs in a task of its own.
+
+        At most one runs per session: if an output flush is requested while one is
+        still sending, it runs once more when that one finishes, so messages stay in
+        order. An error closes this session only. `reactive.flush()` waits for these
+        tasks, so after it returns the messages have been sent.
+        """
+        if self._output_flush_task is not None and not self._output_flush_task.done():
+            self._output_flush_again = True
+            return
+        self._output_flush_task = _reactive_environment._spawn(self._run_output_flush())
+
+    async def _run_output_flush(self) -> None:
+        # `reactive.flush()` from this session's flush callbacks (or tasks they
+        # start) returns right away rather than waiting on this task.
+        _enclosing_run.set(asyncio.current_task())
+        while self._output_flush_enabled:
+            self._output_flush_again = False
+            try:
+                await self._output_flush()
+            except Exception as e:
+                await self._unhandled_error(e)
+            if not self._output_flush_again:
+                return
+
+    async def _output_flush(self) -> None:
+        # Outputs go out once per cycle: while any of this session's effects are
+        # still running, hold everything. The idle transition requests an output flush.
+        if self._busy_count > 0:
+            return
+        if (
+            self._cycle_start_action_queue
+            and not self._cycle_ended
+            and self._outbound_message_queues.is_empty()
+        ):
+            # Nothing to report before the queued action (e.g. an input update) runs.
+            self._start_cycle()
+            return
 
-    async def _flush(self) -> None:
         with session_context(self):
             # This is the only place in the session where the RestoreContext is flushed.
             if self.bookmark._restore_context:
                 self.bookmark._restore_context.flush_pending()
-            # Flush the callbacks
-            await self._flush_callbacks.invoke()
+            # Run the `on_flush` callbacks
+            self._running_flush_callbacks = True
+            try:
+                await self._flush_callbacks.invoke()
+            finally:
+                self._running_flush_callbacks = False
+
+        # Async flush callbacks yield, and an effect may have started meanwhile; its
+        # cycle's idle transition sends everything still queued.
+        if self._busy_count > 0:
+            return
 
         try:
             omq = self._outbound_message_queues
-
-            message: dict[str, object] = {
-                "values": omq.values,
-                "inputMessages": omq.input_messages,
-                "errors": omq.errors,
-            }
-
-            try:
+            # Send only when a cycle has ended or something is queued (R's
+            # `hasPendingUpdates()`). A cycle's message is sent even when empty: the
+            # client uses it to settle outputs still showing progress (e.g. after
+            # `req(False, cancel_output=True)`).
+            if self._cycle_ended or not omq.is_empty():
+                # Take the queued values before awaiting the send, so a value
+                # written during the send stays queued for the next message.
+                message: dict[str, object] = {
+                    "values": dict(omq.values),
+                    "inputMessages": list(omq.input_messages),
+                    "errors": dict(omq.errors),
+                }
+                omq.reset()
+                self._cycle_ended = False
                 await self._send_message(message)
-            finally:
-                self._outbound_message_queues.reset()
         finally:
             with session_context(self):
                 await self._flushed_callbacks.invoke()
 
+        # This cycle's outputs are out; start the next one.
+        self._start_cycle()
+
     def _increment_busy_count(self) -> None:
         self._busy_count += 1
         if self._busy_count == 1:
@@ -1680,6 +1781,36 @@ def _decrement_busy_count(self) -> None:
         self._busy_count -= 1
         if self._busy_count == 0:
             self._send_message_sync({"busy": "idle"})
+            # `_output_flush` sends this cycle's outputs, then starts the next action.
+            self._cycle_ended = True
+            self._request_output_flush()
+
+    def _cycle_start_action(self, action: Callable[[], None]) -> None:
+        """
+        Run `action` at the start of the next cycle, once this session is idle and
+        its outputs have been sent.
+
+        Actions are sync, so they can't be interrupted partway, and only `_output_flush`
+        runs them (never whichever task happened to queue one).
+        """
+        self._cycle_start_action_queue.append(action)
+        self._request_output_flush()
+
+    def _start_cycle(self) -> None:
+        # One action per cycle; the next runs after the outputs this one causes.
+        if self._busy_count != 0 or not self._cycle_start_action_queue:
+            return
+        if self._has_run_session_ended_tasks:
+            self._cycle_start_action_queue.clear()
+            return
+        action = self._cycle_start_action_queue.pop(0)
+        try:
+            action()
+        finally:
+            # The cycle this action starts ends when its effects finish (the idle
+            # transition), or right away if it started none.
+            self._cycle_ended = True
+            self._request_output_flush()
 
     # ==========================================================================
     # On session ended
@@ -1974,6 +2105,9 @@ def _increment_busy_count(self) -> None:
     def _decrement_busy_count(self) -> None:
         self._root_session._decrement_busy_count()
 
+    def _cycle_start_action(self, action: Callable[[], None]) -> None:
+        self._root_session._cycle_start_action(action)
+
     def set_message_handler(
         self,
         name: str,
diff --git a/tests/playwright/shiny/async/concurrent_sessions/app.py b/tests/playwright/shiny/async/concurrent_sessions/app.py
new file mode 100644
index 000000000..b6dafd2cd
--- /dev/null
+++ b/tests/playwright/shiny/async/concurrent_sessions/app.py
@@ -0,0 +1,30 @@
+import asyncio
+
+from shiny import App, Inputs, Outputs, Session, reactive, render, ui
+
+app_ui = ui.page_fluid(
+    ui.input_action_button("block", "Run slow async effect"),
+    ui.output_text("ticks"),
+)
+
+
+def server(input: Inputs, output: Outputs, session: Session) -> None:
+    n = reactive.value(0)
+
+    @reactive.effect
+    def _tick():
+        reactive.invalidate_later(0.1)
+        with reactive.isolate():
+            n.set(n() + 1)
+
+    @reactive.effect
+    @reactive.event(input.block)
+    async def _slow():
+        await asyncio.sleep(3)
+
+    @render.text
+    def ticks():
+        return str(n())
+
+
+app = App(app_ui, server)
diff --git a/tests/playwright/shiny/async/concurrent_sessions/test_concurrent_sessions.py b/tests/playwright/shiny/async/concurrent_sessions/test_concurrent_sessions.py
new file mode 100644
index 000000000..ce4087f9a
--- /dev/null
+++ b/tests/playwright/shiny/async/concurrent_sessions/test_concurrent_sessions.py
@@ -0,0 +1,27 @@
+from playwright.sync_api import Page
+
+from shiny.playwright import controller
+from shiny.run import ShinyAppProc
+
+
+def test_slow_async_effect_does_not_block_other_tab(
+    page: Page, local_app: ShinyAppProc
+) -> None:
+    page.goto(local_app.url)
+    other = page.context.new_page()
+    try:
+        other.goto(local_app.url)
+        ticks = controller.OutputText(other, "ticks")
+        ticks.expect.not_to_have_text("")
+
+        # Start a 3s async effect in the first tab, then check the second tab's
+        # timer keeps ticking while it runs.
+        controller.InputActionButton(page, "block").click()
+        page.wait_for_timeout(300)
+        before = int(ticks.loc.inner_text())
+        other.wait_for_timeout(1000)
+        after = int(ticks.loc.inner_text())
+
+        assert after - before >= 3, f"other tab stalled: {before} -> {after}"
+    finally:
+        other.close()
diff --git a/tests/playwright/shiny/async/streaming_download/app.py b/tests/playwright/shiny/async/streaming_download/app.py
new file mode 100644
index 000000000..879aadf78
--- /dev/null
+++ b/tests/playwright/shiny/async/streaming_download/app.py
@@ -0,0 +1,36 @@
+import asyncio
+from typing import AsyncIterable
+
+from shiny import App, Inputs, Outputs, Session, reactive, render, ui
+
+app_ui = ui.page_fluid(
+    ui.download_button("dl", "Download"),
+    ui.input_action_button("release", "Finish download"),
+    ui.output_text("status"),
+)
+
+
+def server(input: Inputs, output: Outputs, session: Session) -> None:
+    state = reactive.value("idle")
+    release = asyncio.Event()
+
+    @reactive.effect
+    @reactive.event(input.release)
+    def _():
+        release.set()
+
+    @render.download_button(filename="streamed.txt")
+    async def dl() -> AsyncIterable[str]:
+        state.set("streaming")
+        yield "first,"
+        # The stream stays open until the session handles the button click.
+        await release.wait()
+        state.set("done")
+        yield "second"
+
+    @render.text
+    def status():
+        return state()
+
+
+app = App(app_ui, server)
diff --git a/tests/playwright/shiny/async/streaming_download/test_streaming_download.py b/tests/playwright/shiny/async/streaming_download/test_streaming_download.py
new file mode 100644
index 000000000..5d223a16e
--- /dev/null
+++ b/tests/playwright/shiny/async/streaming_download/test_streaming_download.py
@@ -0,0 +1,26 @@
+from pathlib import Path
+
+from playwright.sync_api import Page
+
+from shiny.playwright import controller
+from shiny.run import ShinyAppProc
+
+
+def test_streaming_download_updates_outputs_and_handles_input(
+    page: Page, local_app: ShinyAppProc, tmp_path: Path
+) -> None:
+    page.goto(local_app.url)
+    status = controller.OutputText(page, "status")
+    status.expect_value("idle")
+
+    with page.expect_download() as info:
+        controller.DownloadButton(page, "dl").click()
+        # A value set mid-stream reaches the page while the download is open.
+        status.expect_value("streaming")
+        # The session handles this click while the stream waits on it.
+        controller.InputActionButton(page, "release").click()
+        status.expect_value("done")
+
+    path = tmp_path / "streamed.txt"
+    info.value.save_as(path)
+    assert path.read_text() == "first,second"
diff --git a/tests/pytest/mocktime.py b/tests/pytest/mocktime.py
index 4c9026777..0e1ebd920 100644
--- a/tests/pytest/mocktime.py
+++ b/tests/pytest/mocktime.py
@@ -50,8 +50,10 @@ async def advance_time(self, secs: float):
                 self._time = self._sleepers[0][0]
                 self._sleepers.pop(0)[2].set()
 
-                # Give just-awakened task a chance to run.
-                await yield_event_loop()
+                # Give the just-awakened task, and the reactive flush it requests
+                # (a few loop passes later), a chance to run.
+                for _ in range(10):
+                    await yield_event_loop()
             self._time = end_time
 
             await yield_event_loop()
diff --git a/tests/pytest/test_concurrency.py b/tests/pytest/test_concurrency.py
new file mode 100644
index 000000000..3f0f62e19
--- /dev/null
+++ b/tests/pytest/test_concurrency.py
@@ -0,0 +1,2007 @@
+"""Concurrency behavior of async effects across and within sessions.
+
+Target model (R Shiny's): a flush starts effects and never waits for their async parts.
+A session's busy count defines its cycle; input updates wait until the session is idle,
+while the receive loop and other sessions keep going.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import gc
+import json
+import threading
+from typing import AsyncIterable, Callable, Iterator
+
+import pytest
+from starlette.requests import Request
+
+from shiny import App, Inputs, Outputs, Session, reactive, render, ui
+from shiny._connection import MockConnection
+from shiny.bookmark._bookmark import BookmarkApp
+from shiny.bookmark._restore_state import RestoreContext
+from shiny.reactive._core import ReactiveEnvironment, _reactive_environment
+from shiny.session import get_current_session, session_context
+
+TIMEOUT = 1.0
+
+
+async def wait_until(cond: Callable[[], bool], timeout: float = TIMEOUT) -> bool:
+    async def poll() -> None:
+        while not cond():
+            await asyncio.sleep(0.01)
+
+    try:
+        await asyncio.wait_for(poll(), timeout)
+    except asyncio.TimeoutError:
+        return False
+    return True
+
+
+class Client:
+    """Drives one in-process session over a MockConnection."""
+
+    def __init__(self, server: Callable[[Inputs, Outputs, Session], None]) -> None:
+        self.conn = MockConnection()
+        self.session = App(ui.TagList(), server)._create_session(self.conn)
+        self.task = asyncio.create_task(self.session._run())
+
+    def send(self, msg: dict[str, object]) -> None:
+        self.conn.cause_receive(json.dumps(msg))
+
+    def update(self, **data: object) -> None:
+        self.send({"method": "update", "data": data})
+
+    async def close(self) -> None:
+        self.conn.cause_disconnect()
+        await asyncio.wait_for(self.task, TIMEOUT)
+
+
+@pytest.mark.asyncio
+async def test_slow_effect_does_not_block_other_session():
+    gate = asyncio.Event()
+    a_started = asyncio.Event()
+    b_seen: list[object] = []
+
+    def server_a(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            a_started.set()
+            await gate.wait()
+
+    def server_b(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _x():
+            b_seen.append(input.x())
+
+    a, b = Client(server_a), Client(server_b)
+    try:
+        a.send({"method": "init", "data": {"go": 0}})
+        b.send({"method": "init", "data": {"x": 0}})
+        assert await wait_until(lambda: b_seen == [0])
+
+        a.update(go=1)
+        await asyncio.wait_for(a_started.wait(), TIMEOUT)
+
+        b.update(x=1)
+        assert await wait_until(
+            lambda: b_seen == [0, 1]
+        ), "session B stalled behind session A's slow async effect"
+    finally:
+        gate.set()
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_other_sessions_slow_effect_does_not_block_flush():
+    # A sets a value shared across sessions, which invalidates B's slow effect in the
+    # middle of A's flush. A's flush must not wait for B's effect to finish.
+    shared = reactive.value(0)
+    gate = asyncio.Event()
+    b_started = asyncio.Event()
+    a_seen: list[object] = []
+
+    def server_a(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        def _bump():
+            shared.set(shared() + 1)
+
+        @reactive.effect
+        def _x():
+            a_seen.append(input.x())
+
+    def server_b(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(shared, ignore_init=True)
+        async def _slow():
+            b_started.set()
+            await gate.wait()
+
+    a, b = Client(server_a), Client(server_b)
+    try:
+        b.send({"method": "init", "data": {}})
+        a.send({"method": "init", "data": {"go": 0, "x": 0}})
+        assert await wait_until(lambda: a_seen == [0])
+
+        a.update(go=1)
+        await asyncio.wait_for(b_started.wait(), TIMEOUT)
+
+        a.update(x=1)
+        assert await wait_until(
+            lambda: a_seen == [0, 1]
+        ), "session A stalled behind session B's slow async effect"
+    finally:
+        gate.set()
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_update_while_busy():
+    gate = asyncio.Event()
+    started = asyncio.Event()
+    pinged = asyncio.Event()
+    slow_seen: list[object] = []
+    x_seen: list[object] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            slow_seen.append(input.x())
+            started.set()
+            await gate.wait()
+            slow_seen.append(input.x())
+
+        @reactive.effect
+        def _x():
+            x_seen.append(input.x())
+
+        session.set_message_handler("ping", lambda: pinged.set())
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0, "x": 0}})
+        assert await wait_until(lambda: x_seen == [0])
+
+        c.update(go=1)
+        await asyncio.wait_for(started.wait(), TIMEOUT)
+
+        # The session keeps handling messages while an effect is busy.
+        c.send({"method": "ping", "tag": 1, "args": []})
+        assert await wait_until(
+            pinged.is_set
+        ), "receive loop blocked while an effect was busy"
+
+        # An input update that arrives while busy waits until the session is idle...
+        c.update(x=1)
+        await asyncio.sleep(0.05)
+        assert x_seen == [0]
+
+        # ...so the busy effect sees stable inputs, and the update then applies.
+        gate.set()
+        assert await wait_until(lambda: x_seen == [0, 1])
+        assert slow_seen == [0, 0]
+    finally:
+        gate.set()
+        await c.close()
+
+
+class SlowSendConnection(MockConnection):
+    def __init__(self) -> None:
+        super().__init__()
+        self.sent: list[dict[str, object]] = []
+
+    async def send(self, message: str) -> None:
+        await asyncio.sleep(0)  # transport backpressure
+        self.sent.append(json.loads(message))
+
+
+@pytest.mark.asyncio
+async def test_output_set_during_send_is_kept():
+    conn = SlowSendConnection()
+    session = App(ui.TagList(), None)._create_session(conn)
+    assert isinstance(session.bookmark, BookmarkApp)
+    session.bookmark._set_restore_context(RestoreContext())
+    omq = session._outbound_message_queues
+
+    omq.set_value("x", 1)
+    first = asyncio.create_task(session._output_flush())
+    await asyncio.sleep(0)  # `first` is now awaiting the send
+    omq.set_value("y", 2)
+    await first
+    await session._output_flush()
+
+    assert [m["values"] for m in conn.sent if "values" in m] == [{"x": 1}, {"y": 2}]
+
+
+@pytest.mark.asyncio
+async def test_timer_with_queued_update_does_not_hang():
+    # #2182 hang: when the timer fired during the tick between the session going
+    # idle and its queued update running, the timer ran that update inline, then
+    # cancelled itself mid-update, leaving busy_count stuck at 1.
+    seen: list[object] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _tick():
+            seen.append(input.x())
+            reactive.invalidate_later(0.3)
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"x": 0}})
+        session = c.session
+        # Idle between timer runs (the long timer leaves plenty of room).
+        assert await wait_until(lambda: seen == [0] and session._busy_count == 0)
+
+        # Put the session in that one-tick state by hand: idle, with an update queued.
+        session._cycle_start_action_queue.append(
+            lambda: session._manage_inputs({"x": 1})
+        )
+        await asyncio.sleep(0.35)  # the timer fires
+
+        assert await wait_until(lambda: 1 in seen)
+        n = len(seen)
+        assert await wait_until(lambda: len(seen) > n), "timer effect stopped"
+        c.update(x=2)
+        assert await wait_until(lambda: 2 in seen), "session stopped taking updates"
+    finally:
+        await c.close()
+
+
+# ----------------------------------------------------------------------------
+# Effects and calcs that overlap now that a flush doesn't wait for them
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_async_calc_read_by_two_effects_runs_once():
+    runs = 0
+    src = reactive.value(1)
+
+    @reactive.calc
+    async def data():
+        nonlocal runs
+        runs += 1
+        v = src()
+        await asyncio.sleep(0.02)
+        return v * 10
+
+    seen: list[int] = []
+
+    @reactive.effect
+    async def _e1():
+        seen.append(await data())
+
+    @reactive.effect
+    async def _e2():
+        seen.append(await data())
+
+    await reactive.flush()
+    assert (runs, seen) == (1, [10, 10])
+
+    # The cache still works afterwards.
+    with reactive.isolate():
+        for _ in range(3):
+            assert await data() == 10
+    assert runs == 1
+
+    _e1.destroy()
+    _e2.destroy()
+
+
+@pytest.mark.asyncio
+async def test_effect_runs_do_not_overlap():
+    # A re-run waits for the previous run, so a slower, older run can't finish last.
+    v = reactive.value(1)
+    log: list[tuple[str, int]] = []
+
+    @reactive.effect
+    async def _setter():
+        await asyncio.sleep(0.01)
+        v.set(2)
+
+    @reactive.effect
+    async def _slow_for_old_values():
+        x = v()
+        log.append(("start", x))
+        await asyncio.sleep(0.1 if x == 1 else 0.01)
+        log.append(("end", x))
+
+    await reactive.flush()
+    assert log == [("start", 1), ("end", 1), ("start", 2), ("end", 2)]
+
+    _setter.destroy()
+    _slow_for_old_values.destroy()
+
+
+# ----------------------------------------------------------------------------
+# reactive.flush()
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_flush_from_effect_returns_right_away():
+    # Called from inside an effect run, waiting would wait on itself.
+    done: list[bool] = []
+
+    @reactive.effect
+    async def _e():
+        await asyncio.wait_for(reactive.flush(), TIMEOUT)
+        done.append(True)
+
+    await asyncio.wait_for(reactive.flush(), TIMEOUT)
+    assert done == [True]
+    _e.destroy()
+
+
+@pytest.mark.asyncio
+async def test_flush_from_task_started_by_effect_waits_for_dependents():
+    log: list[str] = []
+    v = reactive.value(0)
+    trigger = reactive.value(0)
+    bg_tasks: list[asyncio.Task[None]] = []
+
+    @reactive.effect
+    def _dep():
+        log.append(f"dep saw {v()}")
+
+    @reactive.effect
+    def _starter():
+        async def bg():
+            await asyncio.sleep(0.01)
+            v.set(1)
+            await reactive.flush()
+            log.append("bg after flush")
+
+        if trigger() == 1:
+            bg_tasks.append(asyncio.create_task(bg()))
+
+    await reactive.flush()
+    trigger.set(1)
+    await reactive.flush()
+    await asyncio.wait_for(bg_tasks[0], TIMEOUT)
+    assert log == ["dep saw 0", "dep saw 1", "bg after flush"]
+
+    _dep.destroy()
+    _starter.destroy()
+
+
+@pytest.mark.asyncio
+async def test_flush_during_active_flush_returns():
+    # With a flushed callback that keeps yielding, reactive.flush() used to keep
+    # re-running rounds and never return.
+    flushes = 0
+
+    async def slow_flushed() -> None:
+        nonlocal flushes
+        flushes += 1
+        await asyncio.sleep(0.01)
+
+    unregister = reactive.on_flushed(slow_flushed)
+    try:
+        v = reactive.value(0)
+
+        @reactive.effect
+        def _e():
+            v()
+
+        await reactive.flush()
+        v.set(1)
+        await asyncio.sleep(0.002)  # the requested round is now mid-callback
+        await asyncio.wait_for(reactive.flush(), TIMEOUT)
+        assert flushes < 10
+        _e.destroy()
+    finally:
+        unregister()
+
+
+@pytest.mark.asyncio
+async def test_extended_task_dependents_see_every_queued_result():
+    async def slow_flushed() -> None:
+        await asyncio.sleep(0.05)
+
+    unregister = reactive.on_flushed(slow_flushed)
+    try:
+
+        @reactive.extended_task
+        async def task(n: int) -> int:
+            await asyncio.sleep(0.02)
+            return n
+
+        seen: list[int] = []
+
+        @reactive.effect
+        def _e():
+            if task.status() == "success":
+                seen.append(task.value())
+
+        task.invoke(1)
+        task.invoke(2)  # queued behind the first
+        await reactive.flush()
+        assert await wait_until(lambda: len(seen) == 2)
+        assert seen == [1, 2]
+        _e.destroy()
+    finally:
+        unregister()
+
+
+# ----------------------------------------------------------------------------
+# Per-session flush requests
+# ----------------------------------------------------------------------------
+
+
+class RecordingConnection(MockConnection):
+    def __init__(self) -> None:
+        super().__init__()
+        self.sent: list[dict[str, object]] = []
+        self.session: "Session | None" = None
+        # The session's busy count each time an output message was sent.
+        self.busy_at_output_send: list[int] = []
+
+    async def send(self, message: str) -> None:
+        msg = json.loads(message)
+        if "values" in msg and self.session is not None:
+            self.busy_at_output_send.append(self.session._busy_count)  # type: ignore
+        self.sent.append(msg)
+
+
+class RecordingClient(Client):
+    def __init__(self, server: Callable[[Inputs, Outputs, Session], None]) -> None:
+        self.recording = RecordingConnection()
+        self.conn = self.recording
+        self.session = App(ui.TagList(), server)._create_session(self.conn)
+        self.recording.session = self.session
+        self.task = asyncio.create_task(self.session._run())
+
+    @property
+    def sent(self) -> list[dict[str, object]]:
+        return self.recording.sent
+
+    def input_messages(self) -> list[object]:
+        return [m for s in self.sent for m in s.get("inputMessages", [])]  # type: ignore
+
+
+@pytest.mark.asyncio
+async def test_input_update_from_background_task_is_sent():
+    c = RecordingClient(lambda input, output, session: None)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: len(c.sent) > 1)
+
+        async def background() -> None:
+            ui.update_text("t", value="hi", session=c.session)
+            await reactive.flush()
+
+        await asyncio.create_task(background())
+        assert await wait_until(lambda: len(c.input_messages()) == 1)
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_message_handler_update_after_await_is_sent():
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def handler() -> None:
+            await asyncio.sleep(0.01)
+            ui.update_text("t", value="hi")
+
+        session.set_message_handler("h", handler)
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        c.send({"method": "h", "tag": 1, "args": []})
+        assert await wait_until(lambda: len(c.input_messages()) == 1)
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_message_handlers_awaiting_flush_do_not_deadlock():
+    v = reactive.value(0)
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def handler() -> int:
+            v.set(v() + 1)
+            await reactive.flush()
+            return 1
+
+        session.set_message_handler("h", handler)
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        c.send({"method": "h", "tag": 1, "args": []})
+        c.send({"method": "h", "tag": 2, "args": []})
+
+        def tags() -> list[object]:
+            return [m["response"]["tag"] for m in c.sent if "response" in m]  # type: ignore
+
+        assert await wait_until(lambda: sorted(tags()) == [1, 2])  # type: ignore
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_one_sessions_flush_error_does_not_stop_others():
+    def server_a(input: Inputs, output: Outputs, session: Session) -> None:
+        calls = 0
+
+        def bad() -> None:
+            # Past the init flush, so the error is raised from a shared flush.
+            nonlocal calls
+            calls += 1
+            if calls > 1:
+                raise RuntimeError("on_flush boom")
+
+        session.on_flush(bad, once=False)
+
+        @reactive.effect
+        def _():
+            input.x()
+
+    def server_b(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _():
+            ui.update_text("t", value=str(input.x()))
+
+    a, b = RecordingClient(server_a), RecordingClient(server_b)
+    try:
+        a.send({"method": "init", "data": {"x": 0}})
+        b.send({"method": "init", "data": {"x": 0}})
+        assert await wait_until(lambda: len(b.input_messages()) == 1)
+        a.update(x=1)
+        # A's error closes A only.
+        assert await wait_until(lambda: a.session._has_run_session_ended_tasks)
+        for i in range(1, 4):
+            b.update(x=i)
+        assert await wait_until(lambda: len(b.input_messages()) == 4)
+        assert not b.session._has_run_session_ended_tasks
+    finally:
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_every_cycle_ends_with_a_message_even_when_empty():
+    # The client settles output progress (e.g. after `req(False,
+    # cancel_output=True)`) on this message, so it's sent with nothing in it.
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _():
+            input.x()
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {"x": 0}})
+
+        await asyncio.sleep(0.05)
+        start = len(c.sent)
+        c.update(x=1)
+
+        def cycle_end_message() -> "dict[str, object] | None":
+            after = c.sent[start:]
+            if {"busy": "idle"} not in after:
+                return None
+            idle = after.index({"busy": "idle"})
+            return next((m for m in after[idle:] if "values" in m), None)
+
+        assert await wait_until(lambda: cycle_end_message() is not None)
+        assert cycle_end_message() == {"values": {}, "inputMessages": [], "errors": {}}
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_queued_actions_dropped_when_session_ends():
+    ran: list[bool] = []
+    c = Client(lambda input, output, session: None)
+    c.send({"method": "init", "data": {}})
+    assert await wait_until(lambda: c.session._output_flush_enabled)
+    await c.close()
+
+    c.session._cycle_start_action_queue.append(lambda: ran.append(True))
+    c.session._start_cycle()
+    assert ran == []
+    assert c.session._cycle_start_action_queue == []
+
+
+@pytest.mark.asyncio
+async def test_only_downloads_make_the_session_busy():
+    session = App(ui.TagList(), None)._create_session(MockConnection())
+    busy: list[int] = []
+
+    async def record(*args: object) -> None:
+        busy.append(session._busy_count)
+
+    session._handle_request_impl = record  # type: ignore
+    for action in ("upload", "dynamic_route", "download"):
+        await session._handle_request(None, action, None)  # type: ignore
+    assert busy == [0, 0, 1]
+
+
+def download_request() -> Request:
+    return Request({"type": "http", "method": "GET", "path": "/", "headers": []})
+
+
+@pytest.mark.parametrize("kind", ["async", "sync"])
+@pytest.mark.asyncio
+async def test_value_set_in_download_handler_updates_outputs_and_effects(kind: str):
+    # https://github.com/posit-dev/py-shiny/issues/1785: a download is a plain HTTP
+    # request, so on main nothing flushed after its handler set a value.
+    notes: list[int] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        count = reactive.value(0)
+
+        if kind == "async":
+
+            @render.download_button(filename="f.txt")
+            async def dl() -> AsyncIterable[str]:
+                count.set(count.get() + 1)
+                yield "data"
+
+        else:
+
+            @render.download_button(filename="f.txt")
+            def dl() -> Iterator[str]:
+                count.set(count.get() + 1)
+                yield "data"
+
+        @render.text
+        def n():
+            return str(count())
+
+        @reactive.effect
+        @reactive.event(count, ignore_init=True)
+        def _():
+            notes.append(count())
+
+    def n_values() -> list[object]:
+        return [m["values"]["n"] for m in c.sent if "n" in m.get("values", {})]  # type: ignore
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {".clientdata_output_n_hidden": False}})
+        assert await wait_until(lambda: n_values()[-1:] == ["0"])
+        for i in (1, 2):
+            response = await c.session._handle_request(
+                download_request(), "download", "dl"
+            )
+            async for _chunk in response.body_iterator:  # type: ignore
+                pass
+            # No client message needed for the output and effect to update.
+            assert await wait_until(lambda i=i: n_values()[-1:] == [str(i)])
+            assert notes == list(range(1, i + 1))
+    finally:
+        await c.close()
+
+
+@pytest.mark.parametrize("kind", ["async", "sync"])
+@pytest.mark.asyncio
+async def test_values_set_mid_stream_are_sent_before_the_stream_ends(kind: str):
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        progress = reactive.value(0)
+
+        if kind == "async":
+
+            @render.download_button(filename="f.txt")
+            async def dl() -> AsyncIterable[str]:
+                for i in (1, 2, 3):
+                    progress.set(i)
+                    yield f"chunk{i}"
+
+        else:
+
+            @render.download_button(filename="f.txt")
+            def dl() -> Iterator[str]:
+                for i in (1, 2, 3):
+                    progress.set(i)
+                    yield f"chunk{i}"
+
+        @render.text
+        def n():
+            return str(progress())
+
+    def n_values() -> list[object]:
+        return [m["values"]["n"] for m in c.sent if "n" in m.get("values", {})]  # type: ignore
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {".clientdata_output_n_hidden": False}})
+        assert await wait_until(lambda: n_values()[-1:] == ["0"])
+        response = await c.session._handle_request(download_request(), "download", "dl")
+        chunks = response.body_iterator.__aiter__()  # type: ignore
+        for i in (1, 2, 3):
+            assert await chunks.__anext__() == f"chunk{i}".encode()
+            # The next chunk isn't read yet, so the stream is still open.
+            assert await wait_until(lambda i=i: n_values()[-1:] == [str(i)])
+        with pytest.raises(StopAsyncIteration):
+            await chunks.__anext__()
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_session_handles_input_while_a_download_streams():
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        release = asyncio.Event()
+
+        @reactive.effect
+        @reactive.event(input.release, ignore_init=True)
+        def _():
+            release.set()
+
+        @render.download_button(filename="f.txt")
+        async def dl() -> AsyncIterable[str]:
+            yield "a"
+            await release.wait()
+            yield "b"
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {"release": 0}})
+        assert await wait_until(lambda: c.session._output_flush_enabled)
+        response = await c.session._handle_request(download_request(), "download", "dl")
+
+        async def read_body() -> bytes:
+            return b"".join([chunk async for chunk in response.body_iterator])  # type: ignore
+
+        body = asyncio.create_task(read_body())
+        await asyncio.sleep(0.05)
+        assert not body.done()
+        # The stream finishes only once the session has handled this input.
+        c.update(release=1)
+        assert await asyncio.wait_for(body, TIMEOUT) == b"ab"
+    finally:
+        await c.close()
+
+
+# ----------------------------------------------------------------------------
+# More edge cases
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_async_calc_invalidated_mid_run_recomputes_for_waiting_reader():
+    src = reactive.value(1)
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    @reactive.calc
+    async def data():
+        v = src()
+        started.set()
+        await release.wait()
+        return v * 10
+
+    async def read() -> int:
+        with reactive.isolate():
+            return await data()
+
+    first = asyncio.create_task(read())
+    await started.wait()
+    second = asyncio.create_task(read())  # waits on the run in flight
+    await asyncio.sleep(0)
+    src.set(2)  # invalidates the run the second reader is waiting on
+    release.set()
+    assert await asyncio.wait_for(first, TIMEOUT) == 10
+    assert await asyncio.wait_for(second, TIMEOUT) == 20
+
+
+@pytest.mark.asyncio
+async def test_async_calc_error_is_shared_by_concurrent_readers():
+    runs = 0
+
+    @reactive.calc
+    async def data():
+        nonlocal runs
+        runs += 1
+        await asyncio.sleep(0.01)
+        raise ValueError("boom")
+
+    async def read() -> str:
+        with reactive.isolate():
+            try:
+                await data()
+            except ValueError as e:
+                return str(e)
+            return "no error"
+
+    results = await asyncio.wait_for(asyncio.gather(read(), read()), TIMEOUT)
+    assert results == ["boom", "boom"]
+    assert runs == 1
+
+
+@pytest.mark.asyncio
+async def test_rerun_of_destroyed_effect_is_skipped():
+    v = reactive.value(0)
+    release = asyncio.Event()
+    runs: list[int] = []
+
+    @reactive.effect
+    async def e():
+        runs.append(v())
+        await release.wait()
+
+    flush_task = asyncio.create_task(reactive.flush())
+    assert await wait_until(lambda: runs == [0])
+    v.set(1)  # queues a re-run behind the one in progress
+    await asyncio.sleep(0.01)
+    e.destroy()
+    release.set()
+    await asyncio.wait_for(flush_task, TIMEOUT)
+    await asyncio.wait_for(reactive.flush(), TIMEOUT)
+    assert runs == [0]
+
+
+@pytest.mark.asyncio
+async def test_priority_orders_effect_starts():
+    order: list[str] = []
+    v = reactive.value(0)
+
+    @reactive.effect(priority=1)
+    async def _low():
+        v()
+        order.append("low start")
+        await asyncio.sleep(0.02)
+        order.append("low end")
+
+    @reactive.effect(priority=10)
+    async def _high():
+        v()
+        order.append("high start")
+        await asyncio.sleep(0.04)
+        order.append("high end")
+
+    await reactive.flush()
+    # Starts follow priority; a lower-priority effect can finish first.
+    assert order == ["high start", "low start", "low end", "high end"]
+    _low.destroy()
+    _high.destroy()
+
+
+@pytest.mark.asyncio
+async def test_many_invalidations_in_one_tick_share_one_round():
+    flushes = 0
+
+    async def count() -> None:
+        nonlocal flushes
+        flushes += 1
+
+    unregister = reactive.on_flushed(count)
+    try:
+        v = reactive.value(0)
+        runs: list[int] = []
+
+        @reactive.effect
+        def _e():
+            runs.append(v())
+
+        await reactive.flush()
+        flushes = 0
+        for i in range(1, 6):
+            v.set(i)
+        assert await wait_until(lambda: runs[-1] == 5)
+        await asyncio.sleep(0.02)
+        assert runs == [0, 5]
+        assert flushes == 1
+        _e.destroy()
+    finally:
+        unregister()
+
+
+@pytest.mark.asyncio
+async def test_flush_from_session_on_flush_callback_returns_right_away():
+    called: list[bool] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def on_flush() -> None:
+            await asyncio.wait_for(reactive.flush(), TIMEOUT)
+            called.append(True)
+
+        session.on_flush(on_flush, once=True)
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: called == [True])
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_input_message_from_on_flush_callback_does_not_loop():
+    # #2449 measured 999 flushes when a repeating on_flush callback queued an input
+    # message that requested another flush.
+    count = 0
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        def on_flush() -> None:
+            nonlocal count
+            count += 1
+            session.send_input_message("t", {"value": count})
+
+        session.on_flush(on_flush, once=False)
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        await asyncio.sleep(0.2)
+        assert count < 5
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_outputs_are_sent_together_once_the_cycle_ends():
+    release = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        def _fast():
+            ui.update_text("fast", value="done")
+
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            await release.wait()
+            ui.update_text("slow", value="done")
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0}})
+        assert await wait_until(lambda: any("values" in m for m in c.sent))
+        c.update(go=1)
+        await asyncio.sleep(0.05)
+        # The fast effect finished, but its output waits for the slow one.
+        assert c.input_messages() == []
+        release.set()
+        assert await wait_until(lambda: len(c.input_messages()) == 2)
+        # ...and both arrive in the same message.
+        batches = [m["inputMessages"] for m in c.sent if m.get("inputMessages")]
+        assert len(batches) == 1
+    finally:
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_timer_waits_while_session_is_busy():
+    release = asyncio.Event()
+    ticks: list[float] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _tick():
+            ticks.append(asyncio.get_running_loop().time())
+            reactive.invalidate_later(0.02)
+
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0}})
+        assert await wait_until(lambda: len(ticks) >= 2)
+        c.update(go=1)
+        assert await wait_until(lambda: c.session._busy_count > 0)
+        n = len(ticks)
+        await asyncio.sleep(0.15)
+        # Timer invalidations wait for the busy cycle, like input changes.
+        assert len(ticks) <= n + 1
+        release.set()
+        assert await wait_until(lambda: len(ticks) > n + 2)
+    finally:
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_message_handler_error_returns_error_response():
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        def handler() -> None:
+            raise RuntimeError("handler boom")
+
+        session.set_message_handler("h", handler)
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        c.send({"method": "h", "tag": 1, "args": []})
+        assert await wait_until(lambda: any("response" in m for m in c.sent))
+        response = next(m["response"] for m in c.sent if "response" in m)
+        assert "handler boom" in str(response)
+        assert not c.session._has_run_session_ended_tasks
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_no_output_flush_requests_after_session_ends():
+    c = Client(lambda input, output, session: None)
+    c.send({"method": "init", "data": {}})
+    assert await wait_until(lambda: c.session._output_flush_enabled)
+    await c.close()
+    c.session.send_input_message("t", {"value": 1})
+    assert c.session.id not in c.session.app._sessions_needing_output_flush
+
+
+@pytest.mark.asyncio
+async def test_session_ending_mid_effect_settles_busy_count():
+    release = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = Client(server)
+    c.send({"method": "init", "data": {"go": 0}})
+    c.update(go=1)
+    assert await wait_until(lambda: c.session._busy_count == 1)
+    await c.close()
+    release.set()
+    assert await wait_until(lambda: c.session._busy_count == 0)
+
+
+@pytest.mark.asyncio
+async def test_flush_waits_for_slow_effects_in_other_sessions():
+    # reactive.flush() means "settled": it also waits for other sessions' effects.
+    release = asyncio.Event()
+    done: list[bool] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            await release.wait()
+            done.append(True)
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0}})
+        c.update(go=1)
+        assert await wait_until(lambda: c.session._busy_count == 1)
+        flush = asyncio.create_task(reactive.flush())
+        await asyncio.sleep(0.05)
+        assert not flush.done()
+        release.set()
+        await asyncio.wait_for(flush, TIMEOUT)
+        assert done == [True]
+    finally:
+        release.set()
+        await c.close()
+
+
+# ----------------------------------------------------------------------------
+# Review round 2
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_flush_through_child_task_of_flush_callback_returns():
+    # `asyncio.gather()` (and `asyncio.wait_for()` before Python 3.12) awaits in a
+    # child task, which used to wait for a flush that couldn't start until the
+    # current one finished.
+    done: list[bool] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def on_flush() -> None:
+            await asyncio.gather(reactive.flush())
+            await asyncio.ensure_future(reactive.flush())
+            done.append(True)
+
+        session.on_flush(on_flush, once=True)
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: done == [True])
+        assert await wait_until(lambda: not _reactive_environment._round_running)
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_flush_through_child_task_of_effect_returns():
+    # The effect queues its own re-run, then flushes via a child task: the child
+    # must not wait for the re-run, which waits for the effect.
+    v = reactive.value(0)
+    log: list[str] = []
+
+    @reactive.effect
+    async def e():
+        n = v()
+        log.append(f"start {n}")
+        if n == 0:
+            v.set(1)
+            await asyncio.gather(reactive.flush())
+        log.append(f"end {n}")
+
+    await asyncio.wait_for(reactive.flush(), TIMEOUT)
+    assert log == ["start 0", "end 0", "start 1", "end 1"]
+    e.destroy()
+
+
+@pytest.mark.asyncio
+async def test_async_calc_invalidated_mid_run_does_not_cache_stale_value():
+    # The stale run is the slower one, so it finishes after the fresh run.
+    v = reactive.value(1)
+    trigger = reactive.value(0)
+    seen: list[tuple[str, int]] = []
+
+    @reactive.calc
+    async def data():
+        x = v()
+        await asyncio.sleep(0.1 if x == 1 else 0.01)
+        return x * 10
+
+    @reactive.effect
+    async def e1():
+        seen.append(("e1", await data()))
+
+    @reactive.effect
+    async def e2():
+        if trigger() == 0:
+            return
+        seen.append(("e2", await data()))
+
+    async def poke() -> None:
+        await asyncio.sleep(0.03)  # data()'s first run is mid-await
+        v.set(2)
+        trigger.set(1)
+
+    poker = asyncio.create_task(poke())
+    await reactive.flush()
+    await poker
+    await asyncio.sleep(0.15)
+    await reactive.flush()
+
+    with reactive.isolate():
+        assert await data() == 20
+    assert ("e2", 20) in seen and seen[-1] == ("e1", 20)
+    e1.destroy()
+    e2.destroy()
+
+
+# ----------------------------------------------------------------------------
+# One test per fix that other tests only cover jointly
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_async_calc_reader_waits_for_the_latest_run():
+    # Run A (x=1) is invalidated mid-run; run B (x=2) starts; a third reader waits
+    # on B. A finishing first must not wake that reader with A's value.
+    v = reactive.value(1)
+
+    @reactive.calc
+    async def data():
+        x = v()
+        await asyncio.sleep(0.05 if x == 1 else 0.15)
+        return x * 10
+
+    async def read() -> int:
+        with reactive.isolate():
+            return await data()
+
+    first = asyncio.create_task(read())
+    await asyncio.sleep(0.01)
+    v.set(2)
+    second = asyncio.create_task(read())
+    await asyncio.sleep(0.01)
+    third = asyncio.create_task(read())
+    assert await asyncio.wait_for(third, TIMEOUT) == 20
+    assert await asyncio.wait_for(second, TIMEOUT) == 20
+    # A's own caller gets A's result; its context was invalidated anyway.
+    assert await asyncio.wait_for(first, TIMEOUT) == 10
+
+
+@pytest.mark.asyncio
+async def test_cancelled_effect_run_ends_busy_period():
+    release = asyncio.Event()
+    running: list[asyncio.Task[object]] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            task = asyncio.current_task()
+            assert task is not None
+            running.append(task)
+            await release.wait()
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0}})
+        c.update(go=1)
+        assert await wait_until(lambda: len(running) == 1)
+        assert c.session._busy_count == 1
+        running[0].cancel()
+        assert await wait_until(lambda: c.session._busy_count == 0)
+    finally:
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_concurrent_rounds_do_not_overlap():
+    active = 0
+    most_active = 0
+
+    async def slow_flushed() -> None:
+        nonlocal active, most_active
+        active += 1
+        most_active = max(most_active, active)
+        await asyncio.sleep(0.02)
+        active -= 1
+
+    unregister = reactive.on_flushed(slow_flushed)
+    try:
+        await asyncio.wait_for(
+            asyncio.gather(
+                _reactive_environment.start_round(),
+                _reactive_environment.start_round(),
+                _reactive_environment.start_round(),
+            ),
+            TIMEOUT,
+        )
+        await asyncio.wait_for(reactive.flush(), TIMEOUT)
+        assert most_active == 1
+    finally:
+        unregister()
+
+
+@pytest.mark.asyncio
+async def test_requested_flush_does_not_inherit_the_requesters_session():
+    seen: list[object] = []
+
+    async def record() -> None:
+        seen.append(get_current_session())
+
+    unregister = reactive.on_flushed(record)
+    try:
+        session = App(ui.TagList(), None)._create_session(MockConnection())
+        v = reactive.value(0)
+
+        @reactive.effect
+        def _e():
+            v()
+
+        await reactive.flush()
+        seen.clear()
+        with session_context(session):
+            v.set(1)  # the flush is requested under this session's context
+        assert await wait_until(lambda: len(seen) > 0)
+        assert seen[0] is None
+        _e.destroy()
+    finally:
+        unregister()
+
+
+@pytest.mark.asyncio
+async def test_on_flush_registered_outside_a_cycle_runs():
+    c = Client(lambda input, output, session: None)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: c.session._output_flush_enabled)
+        await asyncio.sleep(0.02)
+
+        ran: list[bool] = []
+        c.session.on_flush(lambda: ran.append(True))
+        await asyncio.wait_for(reactive.flush(), TIMEOUT)
+        assert await wait_until(lambda: ran == [True])
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_slow_message_handler_does_not_block_other_messages():
+    gate = asyncio.Event()
+    fast_done = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def slow() -> None:
+            await gate.wait()
+
+        session.set_message_handler("slow", slow)
+        session.set_message_handler("fast", lambda: fast_done.set())
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        c.send({"method": "slow", "tag": 1, "args": []})
+        c.send({"method": "fast", "tag": 2, "args": []})
+        assert await wait_until(fast_done.is_set)
+    finally:
+        gate.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_flush_callbacks_wait_while_session_is_busy():
+    release = asyncio.Event()
+    busy_when_called: list[int] = []
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        session.on_flush(
+            lambda: busy_when_called.append(session._busy_count),  # type: ignore
+            once=False,
+        )
+
+        @reactive.effect
+        @reactive.event(input.go, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"go": 0}})
+        c.update(go=1)
+        assert await wait_until(lambda: c.session._busy_count == 1)
+        # Ask for flushes while busy.
+        for _ in range(3):
+            c.session.send_input_message("t", {"value": 1})
+            await asyncio.sleep(0.01)
+        release.set()
+        assert await wait_until(lambda: c.session._busy_count == 0)
+        await asyncio.sleep(0.02)
+        assert busy_when_called and all(b == 0 for b in busy_when_called)
+    finally:
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_no_output_message_while_busy_after_async_flush_callback():
+    # The session becomes busy while an async on_flush callback awaits; the output
+    # message must then wait for the idle transition.
+    shared = reactive.value(0)
+    release = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def on_flush() -> None:
+            await asyncio.sleep(0.02)
+
+        session.on_flush(on_flush, once=False)
+
+        @reactive.effect
+        @reactive.event(shared, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: len(c.recording.busy_at_output_send) == 1)
+        c.session.send_input_message("t", {"value": 1})
+        await asyncio.sleep(0.005)  # the on_flush callback is now awaiting
+        shared.set(1)
+        assert await wait_until(lambda: c.session._busy_count == 1)
+        await asyncio.sleep(0.05)
+        release.set()
+        assert await wait_until(lambda: len(c.input_messages()) == 1)
+        assert all(b == 0 for b in c.recording.busy_at_output_send)
+    finally:
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_queued_update_waits_if_session_turns_busy_during_flushed_callbacks():
+    shared = reactive.value(0)
+    release = asyncio.Event()
+    x_seen: list[object] = []
+    # Once armed, the on_flushed callback signals that it's running, then waits
+    # for the test, so the session can turn busy while it awaits.
+    armed = asyncio.Event()
+    in_callback = asyncio.Event()
+    proceed = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        async def on_flushed() -> None:
+            if armed.is_set():
+                in_callback.set()
+                await proceed.wait()
+
+        session.on_flushed(on_flushed, once=False)
+
+        @reactive.effect
+        def _x():
+            x_seen.append(input.x())
+
+        @reactive.effect
+        @reactive.event(shared, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = Client(server)
+    try:
+        c.send({"method": "init", "data": {"x": 0}})
+        assert await wait_until(lambda: x_seen == [0])
+        await asyncio.sleep(0.05)
+        session = c.session
+        # Something to send plus a queued update: the flush sends, runs its flushed
+        # callbacks, then starts the update.
+        armed.set()
+        session.send_input_message("t", {"value": 1})
+        session._cycle_start_action(lambda: session._manage_inputs({"x": 1}))
+        await asyncio.wait_for(in_callback.wait(), TIMEOUT)
+        shared.set(1)  # the session turns busy while the callback awaits
+        await asyncio.sleep(0.01)
+        proceed.set()
+        await asyncio.sleep(0.05)
+        assert x_seen == [0]
+        release.set()
+        assert await wait_until(lambda: x_seen == [0, 1])
+    finally:
+        proceed.set()
+        release.set()
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_cycle_not_started_by_an_input_still_ends_with_a_message():
+    # A cycle started by a shared value (not a queued action) that queues an
+    # update while busy: its (empty) message must go out before the update runs.
+    shared = reactive.value(0)
+    release = asyncio.Event()
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _x():
+            input.x()
+
+        @reactive.effect
+        @reactive.event(shared, ignore_init=True)
+        async def _slow():
+            await release.wait()
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {"x": 0}})
+
+        await asyncio.sleep(0.05)
+        shared.set(1)
+        assert await wait_until(lambda: c.session._busy_count == 1)
+        c.update(x=1)
+        await asyncio.sleep(0.02)
+        start = len(c.sent)
+        release.set()
+        assert await wait_until(lambda: {"busy": "busy"} in c.sent[start:])
+        after = c.sent[start:]
+        # Between this cycle going idle and the update's cycle starting, the
+        # cycle's message goes out.
+        idle = after.index({"busy": "idle"})
+        busy = after.index({"busy": "busy"})
+        assert any("values" in m for m in after[idle:busy])
+    finally:
+        release.set()
+        await c.close()
+
+
+# ----------------------------------------------------------------------------
+# Each session sends its outputs in its own task
+# ----------------------------------------------------------------------------
+
+
+class GatedConnection(RecordingConnection):
+    """
+    A connection whose output-message sends can be held open (a slow client) or
+    fail. Other messages (e.g. busy/idle status) go straight through.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.gate = asyncio.Event()
+        self.gate.set()
+        self.fail = False
+        self.sending = 0
+        self.most_sending = 0
+
+    async def send(self, message: str) -> None:
+        if '"values"' not in message:
+            await super().send(message)
+            return
+        self.sending += 1
+        self.most_sending = max(self.most_sending, self.sending)
+        try:
+            await self.gate.wait()
+            if self.fail:
+                raise RuntimeError("send failed")
+            await super().send(message)
+        finally:
+            self.sending -= 1
+
+
+class GatedClient(RecordingClient):
+    def __init__(self, server: Callable[[Inputs, Outputs, Session], None]) -> None:
+        self.gated = GatedConnection()
+        self.recording = self.gated
+        self.conn = self.gated
+        self.session = App(ui.TagList(), server)._create_session(self.conn)
+        self.recording.session = self.session
+        self.task = asyncio.create_task(self.session._run())
+
+
+def update_on(x_name: str = "x"):
+    """A server that sends an input message for every change of input `x_name`."""
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _():
+            ui.update_text("t", value=str(input[x_name]()))
+
+    return server
+
+
+async def started(*clients: RecordingClient) -> None:
+    for c in clients:
+        c.send({"method": "init", "data": {"x": 0}})
+    for c in clients:
+        assert await wait_until(lambda c=c: len(c.input_messages()) == 1)
+
+
+def texts(c: RecordingClient) -> list[object]:
+    return [m["message"]["value"] for m in c.input_messages()]  # type: ignore
+
+
+@pytest.mark.asyncio
+async def test_slow_client_does_not_delay_other_sessions_output():
+    a, b = GatedClient(update_on()), GatedClient(update_on())
+    try:
+        await started(a, b)
+        a.gated.gate.clear()  # A's client stops reading
+        a.update(x=1)
+        b.update(x=1)
+        assert await wait_until(lambda: texts(b) == ["0", "1"])
+        assert texts(a) == ["0"]
+        a.gated.gate.set()
+        assert await wait_until(lambda: texts(a) == ["0", "1"])
+    finally:
+        a.gated.gate.set()
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_slow_client_does_not_block_the_next_reactive_flush():
+    shared = reactive.value(0)
+    seen: list[int] = []
+
+    @reactive.effect
+    def _no_session():
+        seen.append(shared())
+
+    a = GatedClient(update_on())
+    try:
+        await started(a)
+        await reactive.flush()
+        a.gated.gate.clear()
+        a.update(x=1)
+        assert await wait_until(lambda: a.gated.sending == 1)
+        # A's send is stuck; unrelated reactive work still runs.
+        shared.set(1)
+        assert await wait_until(lambda: seen[-1] == 1)
+    finally:
+        a.gated.gate.set()
+        _no_session.destroy()
+        await a.close()
+
+
+@pytest.mark.asyncio
+async def test_slow_on_flush_callback_delays_only_its_session():
+    gate = asyncio.Event()
+
+    def slow_server(input: Inputs, output: Outputs, session: Session) -> None:
+        update_on()(input, output, session)
+
+        async def slow() -> None:
+            await gate.wait()
+
+        session.on_flush(slow, once=False)
+
+    a, b = RecordingClient(slow_server), RecordingClient(update_on())
+    try:
+        a.send({"method": "init", "data": {"x": 0}})
+        b.send({"method": "init", "data": {"x": 0}})
+        assert await wait_until(lambda: len(b.input_messages()) == 1)
+        b.update(x=1)
+        assert await wait_until(lambda: texts(b) == ["0", "1"])
+        assert texts(a) == []
+        gate.set()
+        assert await wait_until(lambda: texts(a) == ["0"])
+    finally:
+        gate.set()
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_a_sessions_sends_never_overlap_and_stay_in_order():
+    a = GatedClient(update_on())
+    try:
+        await started(a)
+        a.gated.gate.clear()
+        a.update(x=1)
+        assert await wait_until(lambda: a.gated.sending == 1)
+        # More output queued while the first send is stuck.
+        for i in range(2, 5):
+            a.session.send_input_message("t", {"value": str(i)})
+            await asyncio.sleep(0.01)
+        a.gated.gate.set()
+        assert await wait_until(lambda: texts(a) == ["0", "1", "2", "3", "4"])
+        assert a.gated.most_sending == 1
+        # The requests made while blocked were merged into one more output flush.
+        assert sum("values" in m for m in a.sent[-3:]) <= 2
+    finally:
+        a.gated.gate.set()
+        await a.close()
+
+
+@pytest.mark.asyncio
+async def test_reactive_flush_waits_for_pending_sends():
+    a = GatedClient(update_on())
+    try:
+        await started(a)
+        a.gated.gate.clear()
+        a.session.send_input_message("t", {"value": "hi"})
+        flush = asyncio.create_task(reactive.flush())
+        assert await wait_until(lambda: a.gated.sending == 1)
+        await asyncio.sleep(0.05)
+        assert not flush.done()
+        a.gated.gate.set()
+        await asyncio.wait_for(flush, TIMEOUT)
+        assert texts(a)[-1] == "hi"
+    finally:
+        a.gated.gate.set()
+        await a.close()
+
+
+@pytest.mark.asyncio
+async def test_flush_callbacks_in_two_sessions_calling_reactive_flush_return():
+    # Both sessions flush at once, and each callback awaits reactive.flush(): they
+    # must not wait on each other's flush task.
+    done: list[str] = []
+    a, b = Client(lambda i, o, s: None), Client(lambda i, o, s: None)
+    try:
+        a.send({"method": "init", "data": {}})
+        b.send({"method": "init", "data": {}})
+        assert await wait_until(
+            lambda: a.session._output_flush_enabled and b.session._output_flush_enabled
+        )
+        await asyncio.sleep(0.02)
+
+        def callback(name: str):
+            async def cb() -> None:
+                await reactive.flush()
+                done.append(name)
+
+            return cb
+
+        for name, c in (("a", a), ("b", b)):
+            c.session.on_flush(callback(name), once=True)
+        assert await wait_until(lambda: sorted(done) == ["a", "b"])
+    finally:
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_session_closed_while_its_send_is_stuck():
+    a = GatedClient(update_on())
+    await started(a)
+    a.gated.gate.clear()
+    a.update(x=1)
+    assert await wait_until(lambda: a.gated.sending == 1)
+    # Queued while the send is stuck, so a re-run is pending when the session ends.
+    a.session.send_input_message("t", {"value": "queued"})
+    a.conn.cause_disconnect()
+    assert await wait_until(lambda: not a.session._output_flush_enabled)
+    flush_task = a.session._output_flush_task
+    a.gated.gate.set()
+    assert flush_task is not None
+    await asyncio.wait_for(flush_task, TIMEOUT)
+    # The session is gone: the pending re-run doesn't send.
+    assert "queued" not in texts(a)
+    # ...and later requests don't start an output flush at all.
+    a.session.send_input_message("t", {"value": "late"})
+    await asyncio.sleep(0.02)
+    assert "late" not in texts(a)
+
+
+@pytest.mark.asyncio
+async def test_failed_send_closes_only_that_session():
+    a, b = GatedClient(update_on()), GatedClient(update_on())
+    try:
+        await started(a, b)
+        a.gated.fail = True
+        a.update(x=1)
+        assert await wait_until(lambda: a.session._has_run_session_ended_tasks)
+        b.update(x=1)
+        assert await wait_until(lambda: texts(b) == ["0", "1"])
+        assert not b.session._has_run_session_ended_tasks
+    finally:
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_outputs_stay_in_order_across_cycles_with_a_slow_client():
+    class SlowConnection(GatedConnection):
+        async def send(self, message: str) -> None:
+            if '"values"' in message:
+                await asyncio.sleep(0.005)
+            await super().send(message)
+
+    a = GatedClient(update_on())
+    a.gated.__class__ = SlowConnection
+    try:
+        await started(a)
+        for i in range(1, 6):
+            a.update(x=i)
+        assert await wait_until(lambda: texts(a)[-1:] == ["5"])
+        assert texts(a) == [str(i) for i in range(6)]
+        assert a.gated.most_sending == 1
+    finally:
+        await a.close()
+
+
+# ----------------------------------------------------------------------------
+# Flush requests from other threads
+# ----------------------------------------------------------------------------
+
+
+def in_thread(fn: Callable[[], object]) -> None:
+    t = threading.Thread(target=fn)
+    t.start()
+    t.join(TIMEOUT)
+
+
+@pytest.mark.asyncio
+async def test_value_set_from_another_thread_updates_session_output():
+    shared = reactive.value(0)
+
+    def server(input: Inputs, output: Outputs, session: Session) -> None:
+        @reactive.effect
+        def _():
+            ui.update_text("t", value=str(shared()))
+
+    c = RecordingClient(server)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: texts(c) == ["0"])
+        loop = asyncio.get_running_loop()
+        # Changing reactive state isn't thread-safe; hand the set() to the loop.
+        in_thread(lambda: loop.call_soon_threadsafe(shared.set, 1))
+        assert await wait_until(lambda: texts(c) == ["0", "1"])
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_value_set_from_another_thread_runs_sessionless_effect():
+    # Not the recommended pattern (see `Value.set()`), but with no session involved
+    # it works, and the flush it requests must reach Shiny's loop.
+    v = reactive.value(0)
+    seen: list[int] = []
+
+    @reactive.effect
+    def _e():
+        seen.append(v())
+
+    await reactive.flush()
+    in_thread(lambda: v.set(1))
+    assert await wait_until(lambda: seen == [0, 1])
+    _e.destroy()
+
+
+@pytest.mark.asyncio
+async def test_call_soon_threadsafe_set_from_another_thread():
+    # The documented thread-safe pattern.
+    v = reactive.value(0)
+    seen: list[int] = []
+    loop = asyncio.get_running_loop()
+
+    @reactive.effect
+    def _e():
+        seen.append(v())
+
+    await reactive.flush()
+    in_thread(lambda: loop.call_soon_threadsafe(v.set, 1))
+    assert await wait_until(lambda: seen == [0, 1])
+    _e.destroy()
+
+
+@pytest.mark.asyncio
+async def test_round_request_from_a_thread_with_its_own_loop_goes_to_shinys_loop():
+    main_loop = asyncio.get_running_loop()
+    loops: list[object] = []
+    v = reactive.value(0)
+
+    @reactive.effect
+    def _e():
+        v()
+        loops.append(asyncio.get_running_loop())
+
+    await reactive.flush()
+
+    async def other_loop() -> None:
+        v.set(1)
+
+    in_thread(lambda: asyncio.run(other_loop()))
+    assert await wait_until(lambda: len(loops) == 2)
+    assert loops[1] is main_loop
+    _e.destroy()
+
+
+@pytest.mark.asyncio
+async def test_many_round_requests_from_threads_are_merged():
+    flushes = 0
+
+    async def count() -> None:
+        nonlocal flushes
+        flushes += 1
+
+    await reactive.flush()
+    unregister = reactive.on_flushed(count)
+    try:
+        threads = [
+            threading.Thread(target=_reactive_environment.request_round)
+            for _ in range(20)
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(TIMEOUT)
+        assert await wait_until(lambda: flushes >= 1)
+        await asyncio.sleep(0.05)
+        assert flushes <= 3
+    finally:
+        unregister()
+
+
+def test_round_request_from_a_thread_with_no_loop_yet_is_ignored():
+    env = ReactiveEnvironment()
+    in_thread(env.request_round)
+    assert env._loop is None and not env._round_requested
+
+
+def test_round_request_on_a_new_loop_after_the_old_one_closed():
+    # E.g. `asyncio.run()` twice: the new loop takes over.
+    env = ReactiveEnvironment()
+    flushed: list[bool] = []
+
+    async def mark() -> None:
+        flushed.append(True)
+
+    env.on_round_finished(mark)
+
+    async def use() -> None:
+        env.request_round()
+        await asyncio.sleep(0.01)
+
+    asyncio.run(use())
+    flushed.clear()
+    asyncio.run(use())
+    assert flushed == [True]
+
+
+def test_round_request_after_the_loop_closed_is_ignored():
+    env = ReactiveEnvironment()
+
+    async def use() -> None:
+        env.request_round()
+        await asyncio.sleep(0.01)
+
+    asyncio.run(use())  # env now remembers a loop that is closed
+    assert env._loop is not None and env._loop.is_closed()
+    errors: list[Exception] = []
+
+    def call() -> None:
+        try:
+            env.request_round()
+        except Exception as e:  # pragma: no cover
+            errors.append(e)
+
+    in_thread(call)
+    assert errors == []
+
+
+@pytest.mark.asyncio
+async def test_one_output_message_per_input_update():
+    # No extra empty message when a flush finds nothing to send and no cycle ended.
+    c = RecordingClient(update_on())
+    try:
+        await started(c)
+        await asyncio.sleep(0.05)
+        start = len(c.sent)
+        for i in range(1, 4):
+            c.update(x=i)
+            assert await wait_until(lambda i=i: texts(c)[-1:] == [str(i)])
+            await asyncio.sleep(0.02)
+        assert sum("values" in m for m in c.sent[start:]) == 3
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_on_flush_outside_a_cycle_runs_without_sending_an_empty_message():
+    c = RecordingClient(lambda input, output, session: None)
+    try:
+        c.send({"method": "init", "data": {}})
+        assert await wait_until(lambda: c.session._output_flush_enabled)
+        await asyncio.sleep(0.05)
+        start = len(c.sent)
+        ran: list[bool] = []
+        c.session.on_flush(lambda: ran.append(True))
+        await reactive.flush()
+        assert ran == [True]
+        assert not any("values" in m for m in c.sent[start:])
+    finally:
+        await c.close()
+
+
+def test_round_state_from_a_dead_event_loop_is_discarded():
+    # A round stranded on an event loop that has stopped (e.g. a previous
+    # `test_server()` run, or a test whose loop closed mid-round) used to leave the
+    # environment "in a round" forever: later rounds returned without running, and
+    # `run_next_round()` (used by ExtendedTask) waited forever.
+    env = ReactiveEnvironment()
+    stuck = asyncio.Event()
+
+    async def wait_forever() -> None:
+        await stuck.wait()
+
+    finalized: list[bool] = []
+
+    async def stranded_run() -> None:
+        try:
+            # Nothing else references this future, so only the environment keeps
+            # the task alive.
+            await asyncio.get_running_loop().create_future()
+        finally:
+            # Runs if the task is garbage-collected: in an effect, in the wrong
+            # context, which raised (and failed whichever test was running).
+            finalized.append(True)
+
+    async def strand_a_round() -> None:
+        env._round_finished_callbacks.register(wait_forever)
+        env._spawn(stranded_run())
+        env._spawn(env.start_round())
+        await asyncio.sleep(0.01)  # the round is now waiting on `stuck`
+        assert env._round_running
+        env.request_round()  # also leaves a request pending
+
+    old = asyncio.new_event_loop()
+    old.run_until_complete(strand_a_round())
+    old.close()  # without letting the round finish
+    env._round_finished_callbacks = type(env._round_finished_callbacks)()
+
+    ran: list[bool] = []
+
+    async def record() -> None:
+        ran.append(True)
+
+    async def use_a_new_loop() -> None:
+        env._round_finished_callbacks.register(record)
+        await asyncio.wait_for(env.run_next_round(), TIMEOUT)
+        assert ran == [True]
+        env.request_round()
+        # The dead loop's tasks no longer count as running. (Wait for the round
+        # task's done callback; a timed sleep may not cover it on Windows.)
+        assert await wait_until(lambda: ran == [True, True] and not env._tasks)
+
+    asyncio.run(use_a_new_loop())
+    gc.collect()
+    assert finalized == []  # ...but they're kept, so they're never finalized
diff --git a/tests/pytest/test_otel_async_context.py b/tests/pytest/test_otel_async_context.py
index e8bfcb307..84d37d688 100644
--- a/tests/pytest/test_otel_async_context.py
+++ b/tests/pytest/test_otel_async_context.py
@@ -192,8 +192,8 @@ async def test_parallel_flush_cycles_maintain_context(
 
             # Flush them in parallel
             await asyncio.gather(
-                env1.flush(),
-                env2.flush(),
+                env1.start_round(),
+                env2.start_round(),
             )
 
         # Get exported spans
diff --git a/tests/pytest/test_otel_reactive_flush.py b/tests/pytest/test_otel_reactive_flush.py
index 91e6519fc..4ec4dcd60 100644
--- a/tests/pytest/test_otel_reactive_flush.py
+++ b/tests/pytest/test_otel_reactive_flush.py
@@ -92,7 +92,7 @@ async def test_flush_creates_span_when_collecting(self):
                     "shiny.reactive._core.shiny_otel_span",
                     return_value=mock_span,
                 ) as mock_wrapper:
-                    await env.flush()
+                    await env.start_round()
 
                     # Verify shiny_otel_span was called with correct parameters
                     mock_wrapper.assert_called_once()
@@ -110,7 +110,7 @@ async def test_flush_no_span_when_not_collecting(self):
 
                 # Mock the tracer to verify it's not called at span creation level
                 with patch("shiny.otel._core.get_otel_tracer") as mock_get_tracer:
-                    await env.flush()
+                    await env.start_round()
 
                     # Tracer should not be retrieved since collection level is too low
                     mock_get_tracer.assert_not_called()
@@ -133,7 +133,7 @@ async def test_span_parent_child_relationship(
                 ):
                     # Create reactive environment and flush
                     env = ReactiveEnvironment()
-                    await env.flush()
+                    await env.start_round()
 
         # Get exported spans with proper flushing
         spans = get_exported_spans(provider, memory_exporter)
@@ -180,7 +180,7 @@ async def test_reactive_update_span_includes_session_id_from_context(
             with patch.dict(os.environ, {"SHINY_OTEL_COLLECT": "reactive_update"}):
                 with session_context(mock_session):
                     env = ReactiveEnvironment()
-                    await env.flush()
+                    await env.start_round()
 
         spans = get_exported_spans(provider, memory_exporter)
         app_spans = [s for s in spans if not s.name.startswith("_otel")]
diff --git a/tests/pytest/test_poll.py b/tests/pytest/test_poll.py
index ac6c55257..f985b480c 100644
--- a/tests/pytest/test_poll.py
+++ b/tests/pytest/test_poll.py
@@ -43,6 +43,9 @@ def on_destroy(self, fn: Callable[[], None]) -> Callable[[], None]:
     def _send_message_sync(self, message: Dict[str, object]) -> None:
         pass
 
+    def _cycle_start_action(self, action: Callable[[], None]) -> None:
+        action()
+
     def _increment_busy_count(self) -> None:
         pass
 
diff --git a/tests/pytest/test_reactives.py b/tests/pytest/test_reactives.py
index 22e60ec7f..0915901b6 100644
--- a/tests/pytest/test_reactives.py
+++ b/tests/pytest/test_reactives.py
@@ -303,15 +303,13 @@ async def _():
 
     assert results == [111, 211, 115, 215]
 
-    # This is the order of execution if the async effects are run sequentially. The
-    # `asyncio.sleep(0)` still yields control, but since there are no other effects
-    # scheduled, it will simply resume at the same point.
+    # A flush starts each effect in priority order and doesn't wait for its async
+    # part, so the two effects interleave at each `await`. The order is deterministic
+    # because CPython runs ready callbacks FIFO.
     # fmt: off
     assert exec_order == [
-        'o1-1', 'o1-2', 'r1-1', 'r1-2', 'o1-3',
-        'o2-1', 'o2-2', 'r2-1', 'r2-2', 'o2-3',
-        'o1-1', 'o1-2', 'r1-1', 'r1-2', 'o1-3',
-        'o2-1', 'o2-2', 'r2-1', 'r2-2', 'o2-3'
+        'o1-1', 'o1-2', 'r1-1', 'o2-1', 'r1-2', 'o1-3', 'o2-2', 'r2-1', 'r2-2', 'o2-3',
+        'o1-1', 'o1-2', 'r1-1', 'o2-1', 'r1-2', 'o1-3', 'o2-2', 'r2-1', 'r2-2', 'o2-3',
     ]
     # fmt: on