Streaming callbacks with multiplexed transport

.ai/ARCHITECTURE.md

@@ -1284,6 +1284,207 @@ async def validate_message(websocket, message):
 - `@plotly/dash-websocket-worker/src/worker.ts` - SharedWorker entry point
 - `dash/backends/_fastapi.py` - Server-side WebSocket handler
 
+## Streaming Callbacks
+
+A callback defined as a generator function (or async generator function)
+streams: its yields are pushed to the browser as they are produced — for LLM
+token streaming, progress feeds, and long computations. There is no opt-in
+keyword; `dash._callback.register_callback` infers it from the decorated
+function (`inspect.isgeneratorfunction` / `isasyncgenfunction`) and registers
+the streaming wrapper instead of the regular one.
+
+```python
+import asyncio
+from dash import callback, Output, Input, Patch
+
+@callback(
+    Output('log', 'children'),
+    Input('btn', 'n_clicks'),
+    prevent_initial_call=True,
+)
+async def run(n):
+    yield 'Starting...'          # replaces children immediately
+    async for token in llm():
+        p = Patch()
+        p += token
+        yield p                  # appends to children (incremental)
+    yield 'Done'                 # last yield = final value
+```
+
+### Semantics
+
+- Each yield has the same shape as a regular return value (one value per
+  `Output`) and **replaces** the outputs. Yield `dash.Patch` objects for
+  incremental updates.
+- `no_update` works per-output within a yield; a yield where nothing updates
+  produces no frame. Raising `PreventUpdate` mid-stream ends the stream
+  cleanly.
+- `set_props()` between yields is folded into the next frame's `sideUpdate`
+  (HTTP) or streams immediately (WebSocket transport).
+- Intermediate frames are applied through the same renderer path as
+  `set_props`, so dependent callbacks fire per frame and loading states stay
+  on until the stream completes (`Updating...` title for the whole stream).
+- `on_error` applies per-stream: its return value becomes a final frame.
+  Without it, an exception mid-stream sends an error frame shown in devtools;
+  frames already applied stay applied.
+- The callback must be an `async def` generator on every backend; a
+  synchronous generator is rejected at registration, since it would occupy
+  a server worker (or WS executor thread) for the whole stream.
+- HTTP streams emit a blank keepalive line every `stream_keepalive_interval`
+  ms (`Dash(stream_keepalive_interval=15000)`) that the callback spends
+  between yields, so proxy idle timeouts (nginx `proxy_read_timeout`, 60s by
+  default) don't close a stream mid-thought; `None` disables it. The
+  renderer skips blank lines.
+- Incompatible with `background=True`, `mcp_enabled` and `api_endpoint`
+  (validated at registration, when the function is inspected). Clientside
+  callbacks cannot stream at all.
+- `callback_map[callback_id]['stream']` records the inferred flag server-side;
+  it is not part of the callback spec sent to the client, which detects a
+  stream from the response instead (NDJSON content type / `stream` frames).
+
+### Transport & frame protocol
+
+Transport follows the callback's normal transport selection: if the callback
+runs over the WebSocket callback transport (`websocket=True` or
+`websocket_callbacks=True`), frames ride the open connection as
+`callback_response` messages with `stream: true`; the terminal message is
+`{status: 'ok', stream: true, done: true}`. Otherwise the HTTP POST response
+streams NDJSON (`application/x-ndjson`), one frame per line:
+
+```
+{"multi": true, "response": {"<id>": {"<prop>": <value>}}, "sideUpdate": {...}?}
+{"done": true}                                     <- terminal frame
+{"done": true, "error": {"message": "..."}}        <- error terminal frame
+```
+
+The renderer applies each frame on arrival (via the `sideUpdate` path, so
+`Patch` applies exactly once) and resolves the callback's execution promise
+with an empty result on the terminal frame.
+
+### Multiplexed downlink and the stream SharedWorker
+
+When the app has a shared-storage backend (the default `LocalSharedStorage`),
+HTTP streams do not each hold their own response. The callback's POST carries
+`streamConnection: {requestId}` and returns a fast ack; a *pump*
+(`dash/_stream_hub.py`) drives the generator as an asyncio task and publishes
+each frame, tagged with the request id, to the connection's shared-storage
+topic. The browser's *downlink* (`streamDownlink: {from}`) reads that topic
+and the client routes `{rid, frame, seq}` envelopes back by request id.
+Callback and downlink can be on different workers -- the store is the broker
+-- and a downlink always resumes from its last `seq`, replaying from the
+store's buffer. If the buffer no longer covers the cursor (restart, owner
+re-election) the server sends `{reset: true}` and the client fails its
+in-flight streams instead of silently skipping frames.
+
+The connection id is never chosen by the client: every stream request rides on
+`?endId=`, the server-signed per-page-load token, and the backend derives the
+id from it (`get_stream_connection_id`), answering 403 when it is missing or
+forged -- otherwise a client could read or inject into another page's topic.
+Across worker processes every worker must resolve the same signing secret
+(`secret_key`).
+
+The downlink is hosted in a SharedWorker (`dash-stream-worker.js`, served like
+the WebSocket worker; `config.stream.worker_url`) so **one connection per
+browser** serves every tab: browsers cap HTTP/1.1 connections per host at
+about six, and a downlink per tab stalls the sixth tab. The worker pins the
+`endId` of the tab that opened the downlink while streams are in flight (all
+tabs' frames flow through that one topic). The page talks to the worker
+through `SharedStreamClient` (`utils/streamClient.ts`); the worker runs the
+real `StreamClient` behind `attachStreamWorkerHost`
+(`utils/streamWorkerHost.ts`). Without SharedWorker support the page falls
+back to a downlink of its own.
+
+**Two downlink modes** (`config.stream.mode`, from `backend.downlink_mode`):
+
+- `stream` (ASGI: Quart, FastAPI): one long-lived NDJSON response per
+  browser. It costs no thread -- the subscription parks the task on a future
+  the store resolves (`StoreEngine.apoll`; asyncio streams to the owner from
+  other workers) -- so a single uvicorn worker holds thousands.
+- `poll` (WSGI: Flask): a WSGI response holds a worker thread for its whole
+  life, so an open downlink per browser exhausts a thread pool at a few dozen
+  browsers (gunicorn `--threads 2`: the second browser hung everything).
+  Instead each downlink request returns the frames queued since the cursor
+  and ends at once (`poll_downlink`, `Subscription.poll(0)`), taking a thread
+  for milliseconds. The worker re-polls every `stream_poll_interval` ms
+  (default 100) while frames flow, backs off to five times that after two
+  empty polls (bounding a slow stream's frame latency so frames don't bunch
+  into one poll), and polls immediately when a new stream starts. Pumps are
+  tasks on one event-loop thread per WSGI process (`pump_to_storage`), using
+  the store's loop-native `aget`/`apublish`, not a thread per stream.
+
+**Lifecycle.** Each downlink records its state under the connection's key in
+shared storage: open/closed for a long-lived downlink, a heartbeat (at most
+once a second per worker) for a polling one. Every pump checks it about every
+2s and cancels its callback at its current `await` once the browser is gone:
+a downlink closed for `DOWNLINK_GRACE` (10s), or no poll for `POLL_GRACE`
+(30s -- wide, because an overloaded pool delays polls and overload must cost
+latency, never the stream). A tab closing while other tabs keep the shared
+downlink sends `streamCancel: {requestId}` per stream instead;
+the same pump check picks up the per-request key. A pump that stops publishes
+a terminal `{"done": true}` so a late-reconnecting client resolves.
+
+**Shutdown.** ASGI servers drain in-flight responses before stopping and a
+long-lived downlink never ends by itself, so `_stream_hub` installs a
+SIGINT/SIGTERM handler (`install_stream_shutdown_handler`, at import and again
+from backend startup since uvicorn replaces handlers) that runs
+`shutdown_active_streams` -- sets the shutdown flag, cancels every pump on its
+own loop, closes every open downlink subscription -- then chains to the
+server's own handler. WSGI pumps also stop from an `atexit` hook. The pump
+loop thread shrugs off exceptions raised into it (dash.testing's runner stops
+every thread an app started) and is recreated if it ever dies.
+
+**Scale** (this dev box, 8 cores shared with the load clients; a streaming
+callback per browser yielding every 0.5s; delivery = server yield to client
+receipt):
+
+| server | browsers | frame delay p50 / p95 |
+|---|---|---|
+| uvicorn, 1 worker (FastAPI) | 1000 | 3 ms / 22 ms |
+| uvicorn, 4 workers | 1000 | 1 ms / 3 ms |
+| gunicorn `-w 4 --threads 8` (Flask, poll) | 300 | 105 ms / 200 ms |
+| gunicorn `-w 8 --threads 8` | 1000 | 180 ms / 3.6 s (CPU-bound) |
+| gunicorn `-w 1` (sync worker) | 50 | 50 ms / 100 ms |
+
+Flask works and degrades gracefully -- the cost is a poll per browser per
+interval, so plan roughly one gunicorn worker per 150 concurrently streaming
+browsers -- but for thousands of concurrent streams the ASGI backends are
+the right tool: constant latency and a fraction of the CPU.
+
+### Caveats
+
+- Streaming is inferred from the decorated function, so another decorator
+  between `@callback` and the generator hides it: if that decorator returns a
+  plain function, Dash registers a regular callback and the returned generator
+  object fails to serialize (`InvalidCallbackReturnValue: type generator`).
+- Long streams should check `ctx.websocket.is_shutdown` (WS transport) in
+  loops; on HTTP, client disconnect raises `GeneratorExit` into the user
+  generator at its current `yield`.
+- Proxies and compression middleware (nginx buffering, flask-compress/gzip,
+  Jupyter proxies) can buffer NDJSON and defeat streaming. Dash sets
+  `X-Accel-Buffering: no`, but middleware configuration may still be needed.
+- Streamed frames bypass persistence (`prunePersistence`/`applyPersistence`).
+- Wrapping a streamed output in `dcc.Loading` hides it for the entire stream
+  (loading stays on by design).
+- Flask + async generator requires `dash[async]`; frames are bridged from a
+  private event-loop thread.
+- `flask.request` inside a streamed callback body only works on the pure-WSGI
+  Flask path (no `dash[async]`/`use_async`); under async dispatch the request
+  context cannot be carried into the stream. Use Dash's `ctx` (cookies,
+  headers, args are captured at dispatch) instead.
+
+### Key Files
+
+- `dash/_callback.py` - `add_context_stream`/`async_add_context_stream` wrappers, frame builders
+- `dash/_streaming.py` - `StreamedCallbackResponse` marker, context-safe iteration, NDJSON helpers, keepalives, shutdown flag
+- `dash/_stream_hub.py` - multiplexed transport: `Downlink`, `poll_downlink`, pumps (`pump_to_storage`/`apump_to_storage`), `cancel_stream`, `shutdown_active_streams`/`install_stream_shutdown_handler`
+- `dash/_shared_storage/_engine.py`, `local.py` - `poll`/`apoll`, loop-native `aget`/`aset`/`apublish`, async client connection
+- `dash/backends/_flask.py`, `_quart.py`, `_fastapi.py` - streaming dispatch branches
+- `dash/backends/ws.py` - `make_stream_frame_emitter`, `consume_stream_frames`/`aconsume_stream_frames`
+- `dash/dash-renderer/src/actions/callbacks.ts` - `applyStreamFrame`, NDJSON reader, WS frame handling
+- `dash/dash-renderer/src/utils/workerClient.ts` - stream-aware `callback_response` handling
+- `dash/dash-renderer/src/utils/streamClient.ts` - `StreamClient` (downlink + uplinks), `SharedStreamClient` (page side of the worker), `getStreamClient`
+- `dash/dash-renderer/src/utils/streamWorkerHost.ts`, `src/workers/streamWorker.ts` - the stream SharedWorker
+
 ## Security
 
 ### XSS Protection

.github/workflows/testing.yml

@@ -20,6 +20,7 @@
       dcc_paths_changed: ${{ steps.filter.outputs.dcc_related_paths }}
       html_paths_changed: ${{ steps.filter.outputs.html_related_paths }}
       websocket_changed: ${{ steps.filter.outputs.websocket_paths }}
+      streaming_changed: ${{ steps.filter.outputs.streaming_paths }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -79,6 +80,14 @@
               - '@dash-websocket-worker/**'
               - 'dash/dash-renderer/src/**'
               - 'tests/websocket/**'
+            streaming_paths:
+              - *shared_paths
+              - 'dash/_callback.py'
+              - 'dash/_streaming.py'
+              - 'dash/_callback_context.py'
+              - 'dash/backends/**'
+              - 'dash/dash-renderer/src/**'
+              - 'tests/streaming/**'
 
   lint-unit:
     name: Lint & Unit Tests (Python ${{ matrix.python-version }})
@@ -696,6 +705,69 @@
           touch __init__.py
           pytest --headless --nopercyfinalize tests/websocket -v -s
 
+  streaming-tests:
+    name: Streaming Callback Tests (Python ${{ matrix.python-version }})
+    needs: [build, changes_filter]
+    if: |
+      (github.event_name == 'push' && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/dev')) ||
+      needs.changes_filter.outputs.streaming_changed == 'true'
+    timeout-minutes: 30
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.12"]
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: '24'
+          cache: 'npm'
+
+      - name: Install Node.js dependencies
+        run: npm ci
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: 'pip'
+          cache-dependency-path: requirements/*.txt
+
+      - name: Download built Dash packages
+        uses: actions/download-artifact@v4
+        with:
+          name: dash-packages
+          path: packages/
+
+      - name: Install Dash packages
+        # Streaming callbacks are async generators; the async extra pulls in
+        # flask[async], which they require on the Flask backend.
+        run: |
+          python -m pip install --upgrade pip wheel
+          python -m pip install "setuptools<80.0.0"
+          find packages -name dash-*.whl -print -exec sh -c 'pip install "{}[async,ci,testing,dev,fastapi,quart]"' \;
+
+      - name: Setup Chrome and ChromeDriver
+        uses: browser-actions/setup-chrome@v1
+        with:
+          chrome-version: stable
+
+      - name: Build/Setup test components
+        run: npm run setup-tests.py
+
+      - name: Run streaming tests
+        run: |
+          mkdir streamtests
+          cp -r tests streamtests/tests
+          cd streamtests
+          touch __init__.py
+          pytest --headless --nopercyfinalize tests/streaming -v -s
+
   test-main:
     name: Main Dash Tests (Python ${{ matrix.python-version }}, React ${{ matrix.react-version }}, Group ${{ matrix.test-group }})
     needs: build

CHANGELOG.md

@@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
     - `DiskcacheSharedStorage`: on a `diskcache.Cache`, shared by every process on one host, for single-machine deployments (not for multi-pod ones with ephemeral disks).
     - `RedisSharedStorage`: on Redis, using Redis Streams for the ordered pub/sub: the backend for horizontally-scaled deployments behind a load balancer, e.g. apps scaled across pods.
   - `LocalSharedStorage` additionally accepts `mode=` to make its key/value store durable: `"memory"` (default, in-memory only), `"persist"` (write-through to disk on every change), or `"persist-reset"` (in-memory speed with a periodic flush every `flush_interval` seconds and on clean exit). Persistent modes recover on start and on owner re-election, so state survives a process restart or a crashed owner. Data is stored in a chunked, atomically-written msgpack store (a per-namespace folder under the user cache directory by default, overridable via `path=`). TTLs are preserved across restarts; pub/sub remains transient.
+- [#3931](https://github.com/plotly/dash/pull/3931) Streaming callbacks: a callback defined as an `async def` generator streams its yields to the browser as they are produced (`dash.Patch` yields give incremental updates). All of a browser's streams share one connection hosted in a SharedWorker, so they don't count against the per-host connection limit; closing a tab cancels its streams.
 - [#3977](https://github.com/plotly/dash/pull/3977) Add partial WebSocket prop reads with `get_prop(..., path=...)`. Closes [#3975](https://github.com/plotly/dash/issues/3975).
 - [#3765](https://github.com/plotly/dash/pull/3765) Add opt-in partial pattern matching for callback `Input`, `Output`, and `State` dependencies via `partial_pattern=True`. Dictionary ID patterns can now match component IDs containing additional keys, and partial patterns can be combined with `ALL` and `MATCH` wildcards. Fixes [#3764](https://github.com/plotly/dash/issues/3764).
 - [#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.