Forbid sync-generator streaming callbacks

Sync generator streaming is too limited/buggy (it occupies a server
worker for the whole stream), so reject it at registration with a
StreamCallbackError instead of warning. Async generators are now the
only supported streaming path; remove the dead sync wrapper machinery.

Fix a pre-existing bug on that async path: _astream_frames held one
callback-context token across all yields, but the keepalive driver
resumes each __anext__ in a freshly copied context (via ensure_future),
so the reset ran in a different context and raised. Set/reset the
context var around each generator step (and the on_error handler)
instead. With stream_keepalive_interval enabled by default, this broke
every async streaming callback that used dash.ctx/set_props.

Author: T4rk1n

CHANGELOG.md

@@ -5,7 +5,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Added
-- [#3888](https://github.com/plotly/dash/pull/3888/) Streaming callbacks: a callback defined as a generator (or async generator) function streams — its yields are pushed to the browser as they are produced, no keyword needed. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding `dash.Patch` objects gives incremental updates (e.g. LLM token streaming). Streams ride the WebSocket callback transport when active, otherwise the HTTP response streams NDJSON frames. Works on Flask, Quart, and FastAPI; synchronous generators warn at registration since they occupy a server worker for the whole stream. HTTP streams emit a blank keepalive line every `stream_keepalive_interval` milliseconds (default 15000) that the callback spends between yields, so proxy idle timeouts (nginx's `proxy_read_timeout` defaults to 60s) don't close a stream while the callback is still working; set `stream_keepalive_interval=None` on the app to disable.
+- [#3888](https://github.com/plotly/dash/pull/3888/) Streaming callbacks: a callback defined as a generator (or async generator) function streams — its yields are pushed to the browser as they are produced, no keyword needed. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding `dash.Patch` objects gives incremental updates (e.g. LLM token streaming). Streams ride the WebSocket callback transport when active, otherwise the HTTP response streams NDJSON frames. Works on Flask, Quart, and FastAPI; the callback must be an `async def` generator — synchronous generators are rejected at registration since they occupy a server worker for the whole stream. HTTP streams emit a blank keepalive line every `stream_keepalive_interval` milliseconds (default 15000) that the callback spends between yields, so proxy idle timeouts (nginx's `proxy_read_timeout` defaults to 60s) don't close a stream while the callback is still working; set `stream_keepalive_interval=None` on the app to disable.
 - [#3646](https://github.com/plotly/dash/pull/3646) Experimental support for React 19. The default is still React 18.3.1; to use React 19 set the environment variable `REACT_VERSION=19.2.4` before running your app, or call `dash._dash_renderer._set_react_version("19.2.4")` inside the app. React 19 has no official UMD builds, so Dash serves the [`umd-react`](https://www.npmjs.com/package/umd-react) package, together with a compatibility shim loaded after react-dom and before any component package. The shim keeps component libraries built against React <=18 (e.g. dash-bootstrap-components, dash-mantine-components) working under React 19: it stubs the removed `ReactCurrentOwner` internals, redirects the legacy element `$$typeof` symbol so pre-bundled React 18 jsx-runtimes produce elements React 19 accepts (error #525), and exposes a global `react/jsx-runtime` (`window.ReactJSXRuntime`) that Dash's own component bundles externalize to. Component library authors adopting this convention should copy the defensive `jsxRuntimeExternal` webpack external from `components/dash-core-components/webpack.config.js` rather than a bare `'ReactJSXRuntime'` string: it falls back to a `React.createElement`-based runtime when the global is missing, so the same build also works on Dash versions older than this release.
 
 ### Removed

dash/_callback.py

@@ -3,7 +3,6 @@
 import inspect
 import logging
 import warnings
-from contextvars import copy_context
 from functools import wraps
 from typing import Callable, Optional, Any, List, Tuple, Union, Dict, TypeVar, cast
 
@@ -115,13 +114,14 @@ def callback(
     not to fire when its outputs are first added to the page. Defaults to
     `False` and unlike `app.callback` is not configurable at the app level.
 
-    Decorating a generator function (or async generator function) registers a
-    streaming callback: each yielded value has the same shape as a regular
+    Decorating an async generator function (`async def` with `yield`) registers
+    a streaming callback: each yielded value has the same shape as a regular
     return value (one value per `Output`) and is pushed to the browser
     immediately; yield `dash.Patch` objects for incremental updates. Streams
     over the WebSocket callback transport when active, otherwise over the HTTP
-    response (NDJSON). Streaming callbacks cannot be combined with
-    `background=True`, `mcp_enabled=True` or `api_endpoint`.
+    response (NDJSON). Synchronous generators are not supported (they would
+    occupy a server worker for the whole stream). Streaming callbacks cannot be
+    combined with `background=True`, `mcp_enabled=True` or `api_endpoint`.
 
     :Keyword Arguments:
         :param background:
@@ -270,8 +270,15 @@ def callback(
     )
 
 
-def _validate_stream_callback(callback_id, background, kwargs):
+def _validate_stream_callback(callback_id, background, kwargs, is_sync_gen):
     """Reject options a streaming (generator) callback cannot be combined with."""
+    if is_sync_gen:
+        raise StreamCallbackError(
+            f"Streaming callback '{callback_id}' is a synchronous generator, "
+            "which is not supported: a sync generator occupies a server worker "
+            "for the whole stream. Define it with 'async def' so it streams on "
+            "the event loop instead."
+        )
     if background is not None:
         raise BackgroundCallbackError(
             f"Streaming callback '{callback_id}' cannot be combined with "
@@ -819,7 +826,9 @@ def wrap_func(func):
         is_async_gen_func = inspect.isasyncgenfunction(func)
         is_stream = is_gen_func or is_async_gen_func
         if is_stream:
-            _validate_stream_callback(callback_id, background, _kwargs)
+            _validate_stream_callback(
+                callback_id, background, _kwargs, is_sync_gen=is_gen_func
+            )
 
         if _kwargs.get("api_endpoint"):
             api_endpoint = _kwargs.get("api_endpoint")
@@ -1018,57 +1027,35 @@ def _stream_error_frame(err):
             logger.exception("Exception raised in streamed callback")
             return {"done": True, "error": {"message": str(err) or repr(err)}}
 
-        def _stream_frames(
-            user_gen, error_handler, output_spec, callback_ctx, app, original_packages
-        ):
-            try:
-                while True:
-                    frame = None
-                    try:
-                        output_value = next(user_gen)
-                        frame = _build_stream_frame(
-                            output_value,
-                            output_spec,
-                            callback_ctx,
-                            app,
-                            original_packages,
-                        )
-                    except (StopIteration, PreventUpdate):
-                        break
-                    except Exception as err:  # pylint: disable=broad-exception-caught
-                        if error_handler:
-                            output_value = error_handler(err)
-                            if output_value is not None:
-                                frame = _build_stream_frame(
-                                    output_value,
-                                    output_spec,
-                                    callback_ctx,
-                                    app,
-                                    original_packages,
-                                )
-                                if frame is not None:
-                                    yield frame
-                            break
-                        yield _stream_error_frame(err)
-                        return
-                    if frame is not None:
-                        yield frame
-                yield {"done": True}
-            finally:
-                user_gen.close()
-
         async def _astream_frames(
             user_gen, error_handler, output_spec, callback_ctx, app, original_packages
         ):
-            # The whole stream is iterated from a single task (streaming
-            # response body or WS loop task), so setting the context var here
-            # makes ctx/set_props work for every step of the user generator.
-            token = context_value.set(callback_ctx)
+            # Set the callback context var around each resumption of the user
+            # generator so dash.ctx/set_props resolve while its body runs. It is
+            # set per step rather than once for the whole stream because the
+            # keepalive drivers resume each __anext__ in a freshly copied
+            # context, where a token taken in an earlier step could not be reset.
+            async def _next():
+                token = context_value.set(callback_ctx)
+                try:
+                    return await user_gen.__anext__()
+                finally:
+                    context_value.reset(token)
+
+            def _handle_error(err):
+                # Run the on_error handler under the callback context too so
+                # dash.ctx/set_props resolve inside it, matching a step.
+                token = context_value.set(callback_ctx)
+                try:
+                    return error_handler(err)
+                finally:
+                    context_value.reset(token)
+
             try:
                 while True:
                     frame = None
                     try:
-                        output_value = await user_gen.__anext__()
+                        output_value = await _next()
                         frame = _build_stream_frame(
                             output_value,
                             output_spec,
@@ -1080,7 +1067,7 @@ async def _astream_frames(
                         break
                     except Exception as err:  # pylint: disable=broad-exception-caught
                         if error_handler:
-                            output_value = error_handler(err)
+                            output_value = _handle_error(err)
                             if output_value is not None:
                                 frame = _build_stream_frame(
                                     output_value,
@@ -1098,40 +1085,8 @@ async def _astream_frames(
                         yield frame
                 yield {"done": True}
             finally:
-                context_value.reset(token)
                 await user_gen.aclose()
 
-        @wraps(func)
-        def add_context_stream(*args, **kwargs):
-            """Handles streaming callbacks defined as sync generators."""
-            error_handler = on_error or kwargs.pop("app_on_error", None)
-
-            (
-                output_spec,
-                callback_ctx,
-                func_args,
-                func_kwargs,
-                app,
-                original_packages,
-                _,
-            ) = _initialize_context(
-                args, kwargs, inputs_state_indices, has_output, insert_output
-            )
-
-            # Creates the generator; the function body runs on first next().
-            user_gen = _invoke_callback(func, *func_args, **func_kwargs)
-            frames = _stream_frames(
-                user_gen,
-                error_handler,
-                output_spec,
-                callback_ctx,
-                app,
-                original_packages,
-            )
-            # Snapshot the context (includes the callback context set above) so
-            # the transport can drive the generator after dispatch returns.
-            return StreamedCallbackResponse(frames, is_async=False, ctx=copy_context())
-
         @wraps(func)
         async def async_add_context_stream(*args, **kwargs):
             """Handles streaming callbacks defined as async generators."""
@@ -1161,22 +1116,10 @@ async def async_add_context_stream(*args, **kwargs):
             return StreamedCallbackResponse(frames, is_async=True)
 
         if is_stream:
+            # Only async generators reach here; sync generators are rejected in
+            # _validate_stream_callback above.
             callback_map[callback_id]["stream"] = True
-            if is_gen_func:
-                # A sync generator stream occupies a server worker (or WS
-                # executor thread) for the whole stream duration; recommend
-                # async so it runs on the event loop.
-                warnings.warn(
-                    f"Streaming callback '{callback_id}' is a synchronous "
-                    "generator; it will occupy a server worker for the whole "
-                    "stream. Define it with 'async def' so it runs on the "
-                    "event loop instead.",
-                    RuntimeWarning,
-                    stacklevel=2,
-                )
-                callback_map[callback_id]["callback"] = add_context_stream
-            else:
-                callback_map[callback_id]["callback"] = async_add_context_stream
+            callback_map[callback_id]["callback"] = async_add_context_stream
         elif inspect.iscoroutinefunction(func):
             callback_map[callback_id]["callback"] = async_add_context
         else:

tests/integration/callbacks/test_stream_callbacks.py

@@ -1,8 +1,7 @@
 """Browser integration tests for streaming callbacks over HTTP (NDJSON)."""
+import asyncio
 import time
 
-import pytest
-
 from dash import (
     Dash,
     Input,
@@ -15,7 +14,6 @@
 from dash.testing.wait import until
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst001_stream_progressive_render(dash_duo):
     """Intermediate yields render before the stream completes."""
     app = Dash(__name__)
@@ -31,11 +29,11 @@ def test_stst001_stream_progressive_render(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "step-1"
-        time.sleep(0.5)
+        await asyncio.sleep(0.5)
         yield "step-2"
-        time.sleep(0.5)
+        await asyncio.sleep(0.5)
         yield "done"
 
     dash_duo.start_server(app)
@@ -47,7 +45,6 @@ def stream_cb(n):
     assert dash_duo.get_logs() == []
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst002_stream_patch_appends_once(dash_duo):
     """Patch yields apply exactly once (token streaming)."""
     app = Dash(__name__)
@@ -63,10 +60,10 @@ def test_stst002_stream_patch_appends_once(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "->"
         for token in ["alpha", "beta", "gamma"]:
-            time.sleep(0.2)
+            await asyncio.sleep(0.2)
             patch = Patch()
             patch += token
             yield patch
@@ -81,7 +78,6 @@ def stream_cb(n):
     assert dash_duo.get_logs() == []
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst003_stream_multi_output_and_set_props(dash_duo):
     app = Dash(__name__)
     app.layout = html.Div(
@@ -99,9 +95,9 @@ def test_stst003_stream_multi_output_and_set_props(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "a1", no_update
-        time.sleep(0.3)
+        await asyncio.sleep(0.3)
         set_props("side", {"children": "from-set-props"})
         yield no_update, "b1"
 
@@ -113,7 +109,6 @@ def stream_cb(n):
     assert dash_duo.get_logs() == []
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst004_stream_triggers_downstream_callback(dash_duo):
     """The final streamed value triggers dependent callbacks."""
     app = Dash(__name__)
@@ -130,9 +125,9 @@ def test_stst004_stream_triggers_downstream_callback(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "one"
-        time.sleep(0.2)
+        await asyncio.sleep(0.2)
         yield "two"
 
     @app.callback(
@@ -149,7 +144,6 @@ def downstream(value):
     assert dash_duo.get_logs() == []
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst005_stream_error_shows_in_devtools(dash_duo):
     app = Dash(__name__)
     app.layout = html.Div(
@@ -164,7 +158,7 @@ def test_stst005_stream_error_shows_in_devtools(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "before-error"
         raise ValueError("stream blew up")
 
@@ -176,7 +170,6 @@ def stream_cb(n):
     dash_duo.wait_for_text_to_equal(".test-devtools-error-count", "1")
 
 
-@pytest.mark.filterwarnings("ignore::RuntimeWarning")
 def test_stst006_stream_loading_state(dash_duo):
     """The callback stays in loading state for the whole stream."""
     app = Dash(__name__)
@@ -192,9 +185,9 @@ def test_stst006_stream_loading_state(dash_duo):
         Input("btn", "n_clicks"),
         prevent_initial_call=True,
     )
-    def stream_cb(n):
+    async def stream_cb(n):
         yield "working"
-        time.sleep(1.5)
+        await asyncio.sleep(1.5)
         yield "finished"
 
     dash_duo.start_server(app)