diff --git a/.ai/ARCHITECTURE.md b/.ai/ARCHITECTURE.md
index d90f670f9b..c0208d3944 100644
--- a/.ai/ARCHITECTURE.md
+++ b/.ai/ARCHITECTURE.md
@@ -726,6 +726,7 @@ Special handling for Colab:
 - `prevent_initial_callbacks` - Skip callbacks on load (default: `False`)
 - `background_callback_manager` - DiskcacheManager or CeleryManager
 - `on_error` - Global callback error handler
+- `shared_storage` - Cross-process state + pub/sub backend (default: `LocalSharedStorage`). Pass `None` to disable, or a `BaseSharedStorage` subclass/instance to swap. See [Shared Storage](#shared-storage).
 
 **WebSocket Callbacks:**
 - `websocket_callbacks` - Enable WebSocket for all callbacks (default: `False`). Requires FastAPI backend.
@@ -808,6 +809,183 @@ Supported components: Input, Dropdown, Checklist, RadioItems, Slider, RangeSlide
 | Share state across tabs | `dcc.Store` with `storage_type='local'` |
 | Session-only state | `persistence_type='session'` |
 
+## Shared Storage
+
+Backend-agnostic **server-side** state shared across worker processes: a
+cross-process key/value store plus an ordered, replayable publish/subscribe
+channel. Unlike `dcc.Store` (which lives in the browser), shared storage lives
+on the server and lets callbacks running in different workers see the same keys
+and topics without standing up an external service like Redis.
+
+Enabled by default on every app. It starts lazily on first use, so it costs
+nothing until touched.
+
+### Accessing It
+
+```python
+import dash
+from dash import Input, Output, callback
+
+@callback(Output("out", "children"), Input("btn", "n_clicks"))
+def handler(n):
+    store = dash.ctx.shared_storage      # inside a callback
+    store.set("clicks", n)
+    return store.get("clicks", 0)
+
+# Or off the app directly:
+app = dash.Dash()
+app.shared_storage.set("key", {"any": "json-compatible value"})
+```
+
+Values (and published messages) must be **JSON-compatible** — dict / list / str
+/ int / float / bool / None — the same constraint as `dcc.Store` and callback
+outputs. Messages are encoded with `msgspec` (msgpack), with a stdlib `json`
+fallback.
+
+### API
+
+`app.shared_storage` (and `dash.ctx.shared_storage`) is a `BaseSharedStorage`:
+
+| Method | Purpose |
+|--------|---------|
+| `get(key, default=None)` | Read a value |
+| `set(key, value)` | Write a value |
+| `delete(key)` | Remove a key |
+| `publish(topic, message)` | Append a message to a topic |
+| `subscribe(topic, replay_from=None)` | Return a `Subscription` |
+
+**Ordered pub/sub.** Each `publish` to a topic gets a monotonically increasing
+sequence number (per topic, starting at 1; `0` means "before the first
+message"). A subscriber receives every message published after it subscribed,
+in order:
+
+```python
+# Producer (one worker):
+store.publish("progress", {"pct": 50})
+
+# Consumer (another worker/callback):
+with store.subscribe("progress") as sub:
+    for message in sub:          # or: async for message in sub
+        print(message["pct"])
+```
+
+**Replay on reconnect.** Pass `replay_from=<last-seen-seq>` to resume after a
+drop; buffered messages since that cursor replay first, so a reconnecting
+consumer doesn't miss messages. If the consumer fell farther behind than the
+bounded buffer holds, the subscription raises `SharedStorageGap` (an explicit
+gap rather than a silent hole). A `Subscription` is iterable synchronously
+(`for`) or asynchronously (`async for`), is a context manager, and has
+`close()`.
+
+### Backends
+
+Three backends ship, differing only in **how far state reaches**. Pick by the
+deployment topology — critically, whether the app runs behind a load balancer as
+more than one container/pod (see [Deployment Topology](#deployment-topology-and-shared-storage) below).
+
+| Backend | Backing store | Reaches | Extra |
+|---------|---------------|---------|-------|
+| `LocalSharedStorage` *(default)* | in-memory, owner-elected socket | one **container** (all its workers) | none |
+| `DiskcacheSharedStorage` | a `diskcache.Cache` | one **host** (all processes sharing the dir) | `dash[diskcache]` |
+| `RedisSharedStorage` | Redis (Streams for pub/sub) | **any** process/container/pod | `dash[redis]` |
+
+All three implement the same `BaseSharedStorage` contract (KV + ordered,
+replayable pub/sub with `SharedStorageGap` on buffer overrun), so app code is
+identical across them — only the constructor differs.
+
+```python
+from dash import Dash
+from dash._shared_storage import (
+    LocalSharedStorage, DiskcacheSharedStorage, RedisSharedStorage,
+)
+
+Dash(shared_storage=LocalSharedStorage)                       # default
+Dash(shared_storage=DiskcacheSharedStorage(directory="/tmp/ss"))
+Dash(shared_storage=RedisSharedStorage(url="redis://host:6379"))
+Dash(shared_storage=None)                                     # disabled
+```
+
+When disabled, accessing `app.shared_storage` / `dash.ctx.shared_storage`
+raises `SharedStorageError`.
+
+**`LocalSharedStorage`** — in-memory, no external service. On first use every
+worker races to become the single **owner** for the machine by binding a stable
+address (`AF_UNIX` on POSIX, `127.0.0.1` loopback on Windows); the winner hosts
+the engine, losers proxy over the socket. The bind *is* the lease — when the
+owner dies the address frees and a survivor re-elects **cold** (state is not
+durable). A single-process deployment is its own owner and pays no socket
+overhead. Knobs: `namespace` (defaults to a hash of cwd + argv[0]),
+`buffer_size` (default 2048).
+
+**`DiskcacheSharedStorage`** — KV + an append-log pub/sub on a `diskcache.Cache`,
+the same store `DiskcacheManager` uses; pass an existing `cache` to share one.
+Every process on one host that opens the same directory shares state, so it
+gives multi-worker parity with the local backend without a socket. **Not for
+pods** — each pod has its own ephemeral disk.
+
+**`RedisSharedStorage`** — KV on Redis strings, pub/sub on a Redis Stream per
+topic (an atomic `INCR`+`XADD` script keeps sequences ordered; `MAXLEN` bounds
+the replay window; a subscriber past the trimmed floor gets `SharedStorageGap`).
+Redis is the single source of truth, so no owner election. Pass a `url`
+(defaults to `$REDIS_URL`) or an existing `client` to reuse a connection pool.
+This is the only backend correct for **horizontally-scaled, multi-pod**
+deployments.
+
+### Deployment Topology and Shared Storage
+
+The default `LocalSharedStorage` is **per-container**: its socket/loopback
+election only reaches processes in the same network + filesystem namespace.
+
+- **Single process / single pod** (e.g. Plotly Cloud apps): fine — one owner,
+  nothing to fragment.
+- **Multiple gunicorn workers in one container**: fine — they share the pod's
+  loopback/socket and elect one owner.
+- **Multiple pods behind a load balancer** (e.g. Dash Enterprise apps scaled by
+  an HPA, routed round-robin with no session affinity): **each pod elects its own
+  isolated owner**, so `set()`/`publish()` on one pod are invisible on another.
+  This works at 1 pod and fragments silently once it scales — use
+  `RedisSharedStorage` (one Redis shared by all pods) instead.
+
+### Custom and Out-of-Tree Backends
+
+`BaseSharedStorage` is the stable, public extension point. A backend — shipped
+in-tree or as a **separate package** — implements:
+
+- `get(key, default)` / `set(key, value)` / `delete(key)` — JSON-compatible values
+- `publish(topic, message)` and `subscribe(topic, replay_from=None) -> Subscription`
+- optional `start()` / `close()` (idempotent, called once per worker)
+
+and returns a `Subscription` (`__iter__` / `__aiter__` / `close`) that raises
+`SharedStorageGap` when the replay buffer overran. That trio —
+`BaseSharedStorage`, `Subscription`, `SharedStorageGap` — is exported from
+top-level `dash`; the poll-loop helpers (`PollResult`, the polling subscription)
+are private and not part of the contract.
+
+Core only ships backends whose dependency is already a Dash extra (`msgspec`
+base; `dash[diskcache]`; `dash[redis]`). Anything needing a heavier dependency
+belongs **out of tree** behind this same interface — e.g. a Postgres backend
+(`LISTEN`/`NOTIFY` for push pub/sub + a table for KV and replay) lives in its own
+package so a `psycopg` connection is never pulled into Dash core. It plugs in the
+same way as a built-in: `Dash(shared_storage=PostgresSharedStorage(...))`.
+
+### Module Layout
+
+| File | Responsibility |
+|------|----------------|
+| `_shared_storage/base.py` | Abstract interface: `BaseSharedStorage`, `Subscription`, `SharedStorageError`, `SharedStorageGap` |
+| `_shared_storage/_engine.py` | `StoreEngine`: authoritative in-memory kv map + per-topic ordered log with bounded replay buffer (thread-safe, transport-agnostic) |
+| `_shared_storage/local.py` | `LocalSharedStorage`: owner election; owner hosts the engine, clients proxy |
+| `_shared_storage/diskcache.py` | `DiskcacheSharedStorage`: KV + append-log pub/sub on a `diskcache.Cache` |
+| `_shared_storage/redis.py` | `RedisSharedStorage`: KV + Redis Streams pub/sub (atomic `INCR`+`XADD`) |
+| `_shared_storage/_polling.py` | `PollingSubscription`: shared poll-loop subscription for the diskcache/Redis backends |
+| `_shared_storage/_transport.py` | Length-prefixed, token-gated socket transport (local backend) |
+| `_shared_storage/_codec.py` | msgspec/json codec (data-only) |
+| `dash.py` | `shared_storage` constructor arg + lazy `app.shared_storage` property |
+| `_callback_context.py` | `dash.ctx.shared_storage` accessor |
+
+Requires `msgspec` (in `requirements/install.txt`); `DiskcacheSharedStorage`
+needs the `diskcache` extra and `RedisSharedStorage` the `redis` extra.
+
 ## Async Callbacks
 
 Dash supports `async def` callbacks for non-blocking execution.
diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index 8153a00876..6a90dc94c6 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -242,6 +242,77 @@ jobs:
             tests/compliance/test_typing.py \
             tests/compliance/test_callback_typing.py
 
+  shared-storage:
+    name: Run Shared Storage Tests (Python ${{ matrix.python-version }})
+    needs: build
+    timeout-minutes: 20
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.12"]
+
+    # Redis service for the RedisSharedStorage backend.
+    services:
+      redis:
+        image: redis:6
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    env:
+      REDIS_URL: redis://localhost:6379
+
+    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
+        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 "{}[ci,testing,dev,diskcache,redis]"' \;
+
+      - name: Setup Chrome and ChromeDriver
+        uses: browser-actions/setup-chrome@v1
+        with:
+          chrome-version: stable
+
+      - name: Verify Redis connection
+        run: python -c "import redis; redis.Redis(host='localhost', port=6379).ping(); print('Redis OK')"
+
+      - name: Build/Setup test components
+        run: npm run setup-tests.py
+
+      - name: Run shared-storage tests
+        run: pytest --headless tests/shared_storage -v
+
   background-callbacks:
     name: Run Background & Async Callback Tests (Python ${{ matrix.python-version }})
     needs: [build, changes_filter]
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d3b77ae064..ba43ec2c4e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [#3930](https://github.com/plotly/dash/pull/3930) Shared storage: a backend-agnostic state manager available on every app via `dash.ctx.shared_storage` (and `app.shared_storage`). It offers a cross-process key/value store (`get`/`set`/`delete`) and ordered, replayable publish/subscribe (`publish`/`subscribe`) for sharing state between callbacks or across worker processes without an external service. Subscriptions resume from a caller's last-seen sequence out of a bounded buffer, and a buffer overrun surfaces as an explicit gap rather than a silent loss. Values must be JSON-compatible (like `dcc.Store`). Started lazily on first use, so it costs nothing until touched; pass `shared_storage=None` to `Dash(...)` to disable. Three backends ship, selected by passing a `BaseSharedStorage` instance to `shared_storage=`: the default `LocalSharedStorage` (in-memory, elects one owner process per machine over an `AF_UNIX`/loopback socket — correct within a single container, including across multiple gunicorn workers); `DiskcacheSharedStorage` (on a `diskcache.Cache`, shared by every process on one host — parity for single-machine deployments, not for multi-pod ones with ephemeral disks); and `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).
 - [#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
diff --git a/dash/__init__.py b/dash/__init__.py
index 6f16a068aa..7bd08a47b1 100644
--- a/dash/__init__.py
+++ b/dash/__init__.py
@@ -43,6 +43,16 @@
 from ._patch import Patch  # noqa: F401,E402
 from ._jupyter import jupyter_dash  # noqa: F401,E402
 
+from ._shared_storage import (  # noqa: F401,E402
+    BaseSharedStorage,
+    DiskcacheSharedStorage,
+    LocalSharedStorage,
+    RedisSharedStorage,
+    SharedStorageError,
+    SharedStorageGap,
+    Subscription,
+)
+
 from ._hooks import hooks  # noqa: F401,E402
 
 ctx = callback_context
@@ -94,4 +104,11 @@ def _jupyter_nbextension_paths():
     "ctx",
     "hooks",
     "stringify_id",
+    "BaseSharedStorage",
+    "DiskcacheSharedStorage",
+    "LocalSharedStorage",
+    "RedisSharedStorage",
+    "SharedStorageError",
+    "SharedStorageGap",
+    "Subscription",
 ]
diff --git a/dash/_callback_context.py b/dash/_callback_context.py
index b4285105f5..430d04591a 100644
--- a/dash/_callback_context.py
+++ b/dash/_callback_context.py
@@ -210,6 +210,16 @@ def states_list(self):
     def response(self):
         return getattr(_get_context_value(), "dash_response")
 
+    @property
+    def shared_storage(self):
+        """The app's shared storage (state manager + pub/sub).
+
+        A backend-agnostic key/value + publish/subscribe store shared across
+        worker processes -- for cross-callback or session state without an
+        external service. Requires ``Dash(shared_storage=...)`` (on by default).
+        """
+        return get_app().shared_storage
+
     @staticmethod
     @has_context
     def record_timing(name, duration, description=None):
diff --git a/dash/_shared_storage/__init__.py b/dash/_shared_storage/__init__.py
new file mode 100644
index 0000000000..2dba0ac90c
--- /dev/null
+++ b/dash/_shared_storage/__init__.py
@@ -0,0 +1,28 @@
+"""Backend-agnostic shared state + pub/sub (state manager).
+
+Exposes the ``BaseSharedStorage`` interface and its backends:
+
+- ``LocalSharedStorage`` (default): in-memory, owner-elected -- one container.
+- ``DiskcacheSharedStorage``: on a ``diskcache.Cache`` -- one machine, not pods.
+- ``RedisSharedStorage``: on Redis -- across processes and containers.
+"""
+
+from .base import (
+    BaseSharedStorage,
+    SharedStorageError,
+    SharedStorageGap,
+    Subscription,
+)
+from .diskcache import DiskcacheSharedStorage
+from .local import LocalSharedStorage
+from .redis import RedisSharedStorage
+
+__all__ = [
+    "BaseSharedStorage",
+    "DiskcacheSharedStorage",
+    "LocalSharedStorage",
+    "RedisSharedStorage",
+    "SharedStorageError",
+    "SharedStorageGap",
+    "Subscription",
+]
diff --git a/dash/_shared_storage/_codec.py b/dash/_shared_storage/_codec.py
new file mode 100644
index 0000000000..a683d40756
--- /dev/null
+++ b/dash/_shared_storage/_codec.py
@@ -0,0 +1,37 @@
+"""Codec for the shared-storage socket transport.
+
+Prefers ``msgspec`` (msgpack: fast, compact) and falls back to stdlib ``json``.
+Both are data-only, so bytes read off the socket cannot execute code. All
+workers in one deployment share a single install, hence a single codec, so the
+two ends always agree on the format.
+
+Payloads must be JSON-compatible (dict / list / str / int / float / bool /
+None) under either codec -- the same constraint as ``dcc.Store`` and callback
+outputs.
+"""
+
+from typing import Any
+
+try:
+    import msgspec
+
+    _encoder = msgspec.msgpack.Encoder()
+    _decoder = msgspec.msgpack.Decoder()
+
+    def encode(obj: Any) -> bytes:
+        return _encoder.encode(obj)
+
+    def decode(data: bytes) -> Any:
+        return _decoder.decode(data)
+
+    CODEC = "msgpack"
+except ImportError:  # pragma: no cover - exercised via the fallback install
+    import json
+
+    def encode(obj: Any) -> bytes:
+        return json.dumps(obj).encode("utf-8")
+
+    def decode(data: bytes) -> Any:
+        return json.loads(data.decode("utf-8"))
+
+    CODEC = "json"
diff --git a/dash/_shared_storage/_engine.py b/dash/_shared_storage/_engine.py
new file mode 100644
index 0000000000..7d749b707f
--- /dev/null
+++ b/dash/_shared_storage/_engine.py
@@ -0,0 +1,115 @@
+"""In-memory store engine shared by the owner process and the single-process
+fast path.
+
+Holds the authoritative key/value map and, per topic, an ordered log with a
+bounded replay buffer. Sequence numbers are monotonic per topic starting at 1
+(``0`` means "before the first message"), so a subscriber tracks a cursor and a
+reconnecting one resumes from its last-seen sequence. A consumer that fell
+farther behind than the buffer holds gets an explicit gap signal instead of a
+silent hole.
+
+The engine is thread-safe and transport-agnostic; sockets live one layer up.
+"""
+
+import threading
+import time
+from collections import deque
+from typing import Any, Deque, Dict, List, NamedTuple, Tuple
+
+# Per-topic replay buffer size. Sized to cover a reconnect window comfortably;
+# a producer that outruns a disconnected consumer past this raises a gap rather
+# than dropping messages silently.
+DEFAULT_BUFFER = 2048
+
+
+class PollResult(NamedTuple):
+    messages: List[Any]
+    last_seq: int
+    gap: bool
+
+
+class _Topic:  # pylint: disable=too-few-public-methods
+    __slots__ = ("seq", "buf", "cond")
+
+    def __init__(self, maxlen: int):
+        self.seq = 0
+        self.buf: Deque[Tuple[int, Any]] = deque(maxlen=maxlen)
+        self.cond = threading.Condition()
+
+
+class StoreEngine:
+    def __init__(self, buffer_size: int = DEFAULT_BUFFER):
+        self._buffer_size = buffer_size
+        self._data: Dict[str, Any] = {}
+        self._data_lock = threading.Lock()
+        self._topics: Dict[str, _Topic] = {}
+        self._topics_lock = threading.Lock()
+        self._closed = False
+
+    # --- key/value ---------------------------------------------------------
+    def get(self, key: str, default: Any = None) -> Any:
+        with self._data_lock:
+            return self._data.get(key, default)
+
+    def set(self, key: str, value: Any) -> None:
+        with self._data_lock:
+            self._data[key] = value
+
+    def delete(self, key: str) -> None:
+        with self._data_lock:
+            self._data.pop(key, None)
+
+    # --- pub/sub -----------------------------------------------------------
+    def _topic(self, name: str) -> _Topic:
+        with self._topics_lock:
+            topic = self._topics.get(name)
+            if topic is None:
+                topic = self._topics[name] = _Topic(self._buffer_size)
+            return topic
+
+    def publish(self, topic: str, message: Any) -> int:
+        t = self._topic(topic)
+        with t.cond:
+            t.seq += 1
+            t.buf.append((t.seq, message))
+            t.cond.notify_all()
+            return t.seq
+
+    def head_seq(self, topic: str) -> int:
+        """Current highest sequence -- where a fresh subscription starts."""
+        t = self._topic(topic)
+        with t.cond:
+            return t.seq
+
+    def poll(self, topic: str, after_seq: int, timeout: float) -> PollResult:
+        """Return messages with sequence > ``after_seq``, waiting up to
+        ``timeout`` seconds for at least one. An empty result means the wait
+        elapsed (caller re-polls) or the engine closed. ``gap`` is True when the
+        next expected message was already evicted from the buffer.
+        """
+        t = self._topic(topic)
+        deadline = time.monotonic() + timeout
+        with t.cond:
+            while True:
+                if self._closed:
+                    return PollResult([], after_seq, False)
+                # The next message we want is after_seq + 1; if the buffer's
+                # oldest is newer than that, it was evicted -> gap.
+                if t.buf and after_seq + 1 < t.buf[0][0]:
+                    return PollResult([], after_seq, True)
+                fresh = [m for (s, m) in t.buf if s > after_seq]
+                if fresh:
+                    last = t.buf[-1][0]
+                    return PollResult(fresh, last, False)
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    return PollResult([], after_seq, False)
+                t.cond.wait(remaining)
+
+    def close(self) -> None:
+        self._closed = True
+        with self._topics_lock:
+            topics = list(self._topics.values())
+        for t in topics:
+            with t.cond:
+                t.cond.notify_all()
diff --git a/dash/_shared_storage/_polling.py b/dash/_shared_storage/_polling.py
new file mode 100644
index 0000000000..146b9315b2
--- /dev/null
+++ b/dash/_shared_storage/_polling.py
@@ -0,0 +1,73 @@
+"""Subscription for poll-based backends.
+
+A backend that can answer ``poll(topic, after_seq, timeout) -> PollResult``
+builds its live subscriptions from this: it drives the poll loop, tracks the
+cursor, resumes from it on every call, and turns a buffer overrun into
+``SharedStorageGap``. ``LocalSharedStorage`` has its own reconnecting
+subscription (it also handles owner re-election); this serves the diskcache and
+Redis backends, whose store is always reachable.
+"""
+import asyncio
+import threading
+from typing import Callable
+
+from ._engine import PollResult
+from .base import SharedStorageGap, Subscription
+
+PollFn = Callable[[str, int, float], PollResult]
+
+
+class PollingSubscription(Subscription):
+    """Iterate a topic by repeatedly polling ``poll_fn`` from a moving cursor."""
+
+    def __init__(
+        self, topic: str, start_seq: int, poll_fn: PollFn, poll_timeout: float
+    ):
+        self._topic = topic
+        self._cursor = start_seq
+        self._poll_fn = poll_fn
+        self._poll_timeout = poll_timeout
+        self._closed = threading.Event()
+
+    def close(self) -> None:
+        self._closed.set()
+
+    def _gap(self) -> SharedStorageGap:
+        return SharedStorageGap(f"replay buffer overran on topic {self._topic!r}")
+
+    def __iter__(self):
+        try:
+            while not self._closed.is_set():
+                res = self._poll_fn(self._topic, self._cursor, self._poll_timeout)
+                if res.gap:
+                    raise self._gap()
+                yield from res.messages
+                self._cursor = res.last_seq
+        finally:
+            self.close()
+
+    def __aiter__(self):
+        return self._aiter()
+
+    async def _aiter(self):
+        loop = asyncio.get_running_loop()
+        try:
+            while not self._closed.is_set():
+                try:
+                    res = await loop.run_in_executor(
+                        None,
+                        self._poll_fn,
+                        self._topic,
+                        self._cursor,
+                        self._poll_timeout,
+                    )
+                except RuntimeError:
+                    # The loop/executor is shutting down -- end cleanly.
+                    break
+                if res.gap:
+                    raise self._gap()
+                for message in res.messages:
+                    yield message
+                self._cursor = res.last_seq
+        finally:
+            self.close()
diff --git a/dash/_shared_storage/_transport.py b/dash/_shared_storage/_transport.py
new file mode 100644
index 0000000000..72b0639352
--- /dev/null
+++ b/dash/_shared_storage/_transport.py
@@ -0,0 +1,142 @@
+"""Socket transport for the owner-elected shared store.
+
+A single owner process serves one :class:`StoreEngine` over a stream socket
+(AF_UNIX where available, TCP loopback otherwise). Messages are length-prefixed:
+a 4-byte big-endian length followed by a body encoded by ``_codec`` (msgspec
+msgpack, or stdlib JSON as a fallback). The codec is data-only -- the socket is
+localhost-only, but deserializing untrusted bytes off a socket is a
+remote-code-execution hazard, so the transport carries only data. The channel is
+additionally gated by a per-owner random token; only same-host clients that read
+the owner's advertisement file (written 0600) can attach.
+
+The consequence is that stored values and published messages must be
+JSON-serializable -- the same constraint as ``dcc.Store`` and callback outputs.
+
+Requests are ``[op, *args]`` lists; responses are ``["ok", value]`` or
+``["err", repr]``. Ops: get/set/delete/publish/head/poll -- a thin passthrough
+to the engine. ``poll`` blocks server-side up to its timeout (one thread per
+connection), which is what makes long-poll subscriptions cheap.
+"""
+
+import socket
+import struct
+import threading
+from typing import Any, Optional
+
+from ._codec import decode, encode
+
+EOF = object()  # returned by recv_frame on a clean close
+OK = ["ok", None]
+
+
+def send_frame(sock: socket.socket, obj: Any) -> None:
+    data = encode(obj)
+    sock.sendall(struct.pack("!I", len(data)) + data)
+
+
+def _recv_exactly(sock: socket.socket, n: int) -> Optional[bytes]:
+    chunks = []
+    remaining = n
+    while remaining:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            return None
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
+def recv_frame(sock: socket.socket) -> Any:
+    header = _recv_exactly(sock, 4)
+    if header is None:
+        return EOF
+    (length,) = struct.unpack("!I", header)
+    body = _recv_exactly(sock, length)
+    if body is None:
+        return EOF
+    return decode(body)
+
+
+class OwnerServer:
+    """Serves one StoreEngine to client workers over ``listen_sock``."""
+
+    def __init__(self, engine, listen_sock: socket.socket, token: str):
+        self._engine = engine
+        self._sock = listen_sock
+        self._token = token
+        self._closed = threading.Event()
+        self._accept_thread = threading.Thread(
+            target=self._accept_loop, name="dash-shared-owner", daemon=True
+        )
+
+    def start(self) -> None:
+        self._sock.listen(128)
+        self._accept_thread.start()
+
+    def _accept_loop(self) -> None:
+        while not self._closed.is_set():
+            try:
+                conn, _ = self._sock.accept()
+            except OSError:
+                break
+            threading.Thread(target=self._handle, args=(conn,), daemon=True).start()
+
+    def _handle(self, conn: socket.socket) -> None:
+        try:
+            if recv_frame(conn) != self._token:
+                return
+            send_frame(conn, OK)
+            while not self._closed.is_set():
+                req = recv_frame(conn)
+                if req is EOF:
+                    break
+                send_frame(conn, self._dispatch(req))
+        except (OSError, EOFError, ValueError):
+            pass
+        finally:
+            try:
+                conn.close()
+            except OSError:
+                pass
+
+    def _dispatch(self, req):  # pylint: disable=too-many-return-statements
+        op = req[0]
+        try:
+            if op == "get":
+                return ("ok", self._engine.get(req[1], req[2]))
+            if op == "set":
+                self._engine.set(req[1], req[2])
+                return ("ok", None)
+            if op == "delete":
+                self._engine.delete(req[1])
+                return ("ok", None)
+            if op == "publish":
+                return ("ok", self._engine.publish(req[1], req[2]))
+            if op == "head":
+                return ("ok", self._engine.head_seq(req[1]))
+            if op == "poll":
+                return ("ok", self._engine.poll(req[1], req[2], req[3]))
+            return ("err", f"unknown op {op!r}")
+        except Exception as err:  # pylint: disable=broad-exception-caught
+            return ("err", repr(err))
+
+    def close(self) -> None:
+        self._closed.set()
+        try:
+            self._sock.close()
+        except OSError:
+            pass
+        self._engine.close()
+
+
+def connect_to_owner(family: int, address, token: str, timeout: float = 5.0):
+    """Open a client connection and complete the token handshake."""
+    sock = socket.socket(family, socket.SOCK_STREAM)
+    sock.settimeout(timeout)
+    sock.connect(address)
+    send_frame(sock, token)
+    if recv_frame(sock) != OK:
+        sock.close()
+        raise ConnectionError("shared-storage owner rejected the handshake")
+    sock.settimeout(None)
+    return sock
diff --git a/dash/_shared_storage/base.py b/dash/_shared_storage/base.py
new file mode 100644
index 0000000000..7b7e8699ed
--- /dev/null
+++ b/dash/_shared_storage/base.py
@@ -0,0 +1,114 @@
+"""Backend-agnostic shared state + pub/sub for Dash.
+
+A ``BaseSharedStorage`` gives every worker process a common key/value store and
+an ordered publish/subscribe channel. Dash uses it internally, and apps can use
+it directly for cross-callback or session state without reaching for an external
+service.
+
+Semantics every backend must honor:
+
+- **Values are JSON-compatible.** They may cross a process boundary, the same
+  constraint as ``dcc.Store`` and callback outputs.
+- **Pub/sub is ordered and replayable.** Each ``publish`` to a topic gets a
+  monotonically increasing sequence number; a subscriber receives every message
+  published after it subscribed, in order. A consumer that drops and resubscribes
+  replays what it missed from a bounded buffer, so no message is silently lost
+  within that window -- a buffer overrun surfaces as an explicit gap error rather
+  than a missing message.
+"""
+
+import abc
+from typing import Any, AsyncIterator, Iterator, Optional
+
+
+class SharedStorageError(Exception):
+    """An operation failed on the shared-storage backend."""
+
+
+class SharedStorageGap(SharedStorageError):
+    """Raised by a subscription when messages were evicted before delivery.
+
+    Signals that the replay buffer overran (the consumer fell too far behind),
+    or that the owner was re-elected and its buffer was lost; the caller must
+    treat it as lost data rather than a clean end of the subscription.
+    """
+
+
+class Subscription(abc.ABC):
+    """A live, ordered view of a topic.
+
+    Iterate it synchronously (``for msg in sub``) or asynchronously
+    (``async for msg in sub``) to receive messages as they are published;
+    messages buffered since the subscription's cursor replay first. Iteration
+    ends when the subscription is closed. Raises ``SharedStorageGap`` if the
+    buffer overran while the consumer was behind.
+    """
+
+    @abc.abstractmethod
+    def __iter__(self) -> Iterator[Any]:
+        ...
+
+    @abc.abstractmethod
+    def __aiter__(self) -> AsyncIterator[Any]:
+        ...
+
+    @abc.abstractmethod
+    def close(self) -> None:
+        ...
+
+    def __enter__(self) -> "Subscription":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+    async def __aenter__(self) -> "Subscription":
+        return self
+
+    async def __aexit__(self, *exc) -> None:
+        self.close()
+
+
+class BaseSharedStorage(abc.ABC):
+    """Shared key/value store + ordered pub/sub, usable across worker processes.
+
+    A backend must be safe to construct in every worker. However the authoritative
+    state is held (a single elected owner, an external service, ...), all workers
+    that share a backend see the same keys and topics.
+    """
+
+    def start(self) -> None:
+        """Prepare the backend for use (elect/attach to the owner, connect, ...).
+
+        Called once per worker before first use. Idempotent.
+        """
+
+    def close(self) -> None:
+        """Release this worker's handle on the backend. Idempotent."""
+
+    @abc.abstractmethod
+    def get(self, key: str, default: Any = None) -> Any:
+        ...
+
+    @abc.abstractmethod
+    def set(self, key: str, value: Any) -> None:
+        ...
+
+    @abc.abstractmethod
+    def delete(self, key: str) -> None:
+        ...
+
+    @abc.abstractmethod
+    def publish(self, topic: str, message: Any) -> None:
+        """Append ``message`` to ``topic``; delivered to every current subscriber."""
+
+    @abc.abstractmethod
+    def subscribe(self, topic: str, replay_from: Optional[int] = None) -> Subscription:
+        """Subscribe to ``topic``.
+
+        By default the subscription starts at the topic's current head, so it
+        only receives messages published from now on. ``replay_from`` (a sequence
+        number a previous subscription last saw) resumes after that point,
+        replaying buffered messages -- this is how a reconnecting consumer avoids
+        losing messages.
+        """
diff --git a/dash/_shared_storage/diskcache.py b/dash/_shared_storage/diskcache.py
new file mode 100644
index 0000000000..72980b4fd7
--- /dev/null
+++ b/dash/_shared_storage/diskcache.py
@@ -0,0 +1,135 @@
+"""Disk-backed shared storage (diskcache), for single-machine multi-process use.
+
+Key/value and ordered pub/sub on a ``diskcache.Cache`` -- the same store the
+``DiskcacheManager`` background-callback backend uses. Every process on one host
+that opens the same cache directory shares state, so it gives multi-worker
+parity with ``LocalSharedStorage`` without a socket owner.
+
+It is NOT for multi-container / multi-pod deployments: each pod has its own
+ephemeral disk, so pods would not see each other's state. Use
+``RedisSharedStorage`` there.
+
+Pub/sub is an append log: an atomic per-topic counter assigns monotonic
+sequence numbers, each message is stored under its sequence and old sequences
+are trimmed to a bounded window, and subscribers poll for sequences past their
+cursor. A consumer that falls farther behind than the window gets a
+``SharedStorageGap``.
+"""
+import time
+from typing import Any, Optional
+
+from ._codec import decode, encode
+from ._engine import DEFAULT_BUFFER, PollResult
+from ._polling import PollingSubscription
+from .base import BaseSharedStorage, Subscription
+
+# Poll cycle: short so a subscription's close() stays responsive; diskcache has
+# no server-side blocking wait, so this is a sleep-poll loop.
+_POLL_TIMEOUT = 1.0
+_POLL_INTERVAL = 0.05
+
+
+def _require_diskcache():
+    try:
+        import diskcache  # type: ignore[import-not-found,import-untyped] # pylint: disable=import-outside-toplevel
+
+        return diskcache
+    except ImportError as exc:
+        raise ImportError(
+            "DiskcacheSharedStorage requires the diskcache extra:\n\n"
+            '    $ pip install "dash[diskcache]"\n'
+        ) from exc
+
+
+class DiskcacheSharedStorage(BaseSharedStorage):
+    """Shared storage backed by a ``diskcache.Cache``. Single machine, not pods.
+
+    Pass an existing ``cache`` (e.g. the one a ``DiskcacheManager`` already
+    holds) to share one store, or a ``directory`` to open/create one. Values and
+    published messages must be JSON-compatible.
+    """
+
+    def __init__(
+        self,
+        cache: Any = None,
+        directory: Optional[str] = None,
+        buffer_size: int = DEFAULT_BUFFER,
+    ):
+        diskcache = _require_diskcache()
+        if cache is not None:
+            if not isinstance(cache, (diskcache.Cache, diskcache.FanoutCache)):
+                raise ValueError(
+                    "cache must be a diskcache.Cache or diskcache.FanoutCache"
+                )
+            self._cache = cache
+            self._owns_cache = False
+        else:
+            self._cache = diskcache.Cache(directory)
+            self._owns_cache = True
+        self._buffer_size = buffer_size
+
+    def close(self) -> None:
+        if self._owns_cache:
+            self._cache.close()
+
+    # --- key layout --------------------------------------------------------
+    @staticmethod
+    def _kv(key: str) -> str:
+        return f"ss:kv:{key}"
+
+    @staticmethod
+    def _seq(topic: str) -> str:
+        return f"ss:seq:{topic}"
+
+    @staticmethod
+    def _msg(topic: str, seq: int) -> str:
+        return f"ss:msg:{topic}:{seq}"
+
+    # --- key/value ---------------------------------------------------------
+    def get(self, key: str, default: Any = None) -> Any:
+        raw = self._cache.get(self._kv(key))
+        return default if raw is None else decode(raw)
+
+    def set(self, key: str, value: Any) -> None:
+        self._cache.set(self._kv(key), encode(value))
+
+    def delete(self, key: str) -> None:
+        self._cache.delete(self._kv(key))
+
+    # --- pub/sub -----------------------------------------------------------
+    def publish(self, topic: str, message: Any) -> None:
+        payload = encode(message)
+        with self._cache.transact():
+            seq = int(self._cache.incr(self._seq(topic)))
+            self._cache.set(self._msg(topic, seq), payload)
+            evicted = seq - self._buffer_size
+            if evicted >= 1:
+                self._cache.delete(self._msg(topic, evicted))
+
+    def _head(self, topic: str) -> int:
+        return int(self._cache.get(self._seq(topic), 0))
+
+    def _poll(self, topic: str, after_seq: int, timeout: float) -> PollResult:
+        deadline = time.monotonic() + timeout
+        while True:
+            head = self._head(topic)
+            if head > after_seq:
+                floor = max(1, head - self._buffer_size + 1)
+                if after_seq + 1 < floor:
+                    return PollResult([], after_seq, True)
+                messages = []
+                for seq in range(after_seq + 1, head + 1):
+                    raw = self._cache.get(self._msg(topic, seq))
+                    if raw is None:
+                        # Trimmed between the head read and this fetch -> gap.
+                        return PollResult([], after_seq, True)
+                    messages.append(decode(raw))
+                return PollResult(messages, head, False)
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return PollResult([], after_seq, False)
+            time.sleep(min(_POLL_INTERVAL, remaining))
+
+    def subscribe(self, topic: str, replay_from: Optional[int] = None) -> Subscription:
+        start = replay_from if replay_from is not None else self._head(topic)
+        return PollingSubscription(topic, start, self._poll, _POLL_TIMEOUT)
diff --git a/dash/_shared_storage/local.py b/dash/_shared_storage/local.py
new file mode 100644
index 0000000000..fc0618fdea
--- /dev/null
+++ b/dash/_shared_storage/local.py
@@ -0,0 +1,401 @@
+"""Owner-elected, cross-process shared storage.
+
+Every worker constructs a :class:`LocalSharedStorage`; on first use they race to
+become the single *owner* by binding a stable address (AF_UNIX socket on POSIX,
+TCP loopback on Windows). The winner hosts the authoritative :class:`StoreEngine`
+and serves it; the losers become clients that connect to it. The bind itself is
+the lease -- when the owner dies the address frees, and a client that finds it
+gone re-elects (coming up cold, by design for the in-memory backend).
+
+Because the owner holds the engine directly, a single-process deployment (its
+own owner) pays no socket overhead at all; only extra worker processes proxy.
+"""
+
+import asyncio
+import atexit
+import hashlib
+import json
+import os
+import secrets
+import socket
+import sys
+import tempfile
+import threading
+import time
+from typing import Any, Optional
+
+from ._engine import DEFAULT_BUFFER, PollResult, StoreEngine
+from ._transport import EOF, OwnerServer, connect_to_owner, recv_frame, send_frame
+from .base import BaseSharedStorage, SharedStorageError, SharedStorageGap, Subscription
+
+_HAS_AF_UNIX = hasattr(socket, "AF_UNIX")
+_CLIENT_POLL_TIMEOUT = 20.0  # long-poll cycle for remote subscribers
+_OWNER_POLL_TIMEOUT = 1.0  # local poll cycle; short so close() stays responsive
+
+
+def _default_namespace() -> str:
+    raw = f"{os.getcwd()}|{sys.argv[0] if sys.argv else ''}".encode()
+    return hashlib.sha1(raw).hexdigest()[:16]
+
+
+def _paths(namespace: str):
+    base = os.path.join(tempfile.gettempdir(), f"dash-shared-{namespace}")
+    return base + ".sock", base + ".addr"
+
+
+def _tcp_port(namespace: str) -> int:
+    h = int(hashlib.sha1(namespace.encode()).hexdigest(), 16)
+    return 49152 + (h % (65535 - 49152))
+
+
+def _write_advertisement(addr_path, family, address, token) -> None:
+    payload = json.dumps(
+        {"family": int(family), "address": address, "token": token}
+    ).encode("utf-8")
+    tmp = f"{addr_path}.{os.getpid()}.tmp"
+    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "wb") as f:
+        f.write(payload)
+    os.replace(tmp, addr_path)
+
+
+def _read_advertisement(addr_path, retries=100, delay=0.05):
+    for _ in range(retries):
+        try:
+            with open(addr_path, "rb") as f:
+                data = json.loads(f.read().decode("utf-8"))
+            family = data["family"]
+            address = data["address"]
+            if family == int(socket.AF_INET):
+                address = tuple(address)  # json list -> (host, port)
+            return family, address, data["token"]
+        except (FileNotFoundError, ValueError, KeyError):
+            time.sleep(delay)
+    raise ConnectionError("shared-storage owner advertisement not found")
+
+
+def _unix_is_stale(path: str) -> bool:
+    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    probe.settimeout(0.5)
+    try:
+        probe.connect(path)
+        return False  # a live owner answered
+    except OSError:
+        return True  # refused / no listener -> stale socket file
+    finally:
+        probe.close()
+
+
+def _try_bind(namespace: str, sock_path: str):
+    """Try to become the owner. Returns (listen_sock, family, address) or None."""
+    if _HAS_AF_UNIX:
+        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        try:
+            s.bind(sock_path)
+        except OSError:
+            if _unix_is_stale(sock_path):
+                try:
+                    os.unlink(sock_path)
+                    s.bind(sock_path)
+                except OSError:
+                    s.close()
+                    return None
+            else:
+                s.close()
+                return None
+        os.chmod(sock_path, 0o600)
+        return s, socket.AF_UNIX, sock_path
+
+    port = _tcp_port(namespace)
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    try:
+        s.bind(("127.0.0.1", port))
+    except OSError:
+        s.close()
+        return None
+    return s, socket.AF_INET, ("127.0.0.1", port)
+
+
+class _Coordinator:
+    """Owns this worker's role (owner or client) and re-elects on owner loss."""
+
+    def __init__(self, namespace: str, buffer_size: int):
+        self._namespace = namespace
+        self._buffer_size = buffer_size
+        self._sock_path, self._addr_path = _paths(namespace)
+        self._lock = threading.RLock()
+        self._role: Optional[str] = None
+        self._engine: Optional[StoreEngine] = None
+        self._server: Optional[OwnerServer] = None
+        self._family = None
+        self._address = None
+        self._token: Optional[str] = None
+
+    @property
+    def engine(self) -> Optional[StoreEngine]:
+        return self._engine
+
+    @property
+    def token(self) -> Optional[str]:
+        return self._token
+
+    def ensure(self) -> None:
+        if self._role is None:
+            with self._lock:
+                if self._role is None:
+                    self._elect()
+
+    def is_owner(self) -> bool:
+        self.ensure()
+        return self._role == "owner"
+
+    def _elect(self) -> None:
+        bound = _try_bind(self._namespace, self._sock_path)
+        if bound:
+            listen_sock, family, address = bound
+            token = secrets.token_hex(16)
+            _write_advertisement(self._addr_path, family, address, token)
+            engine = StoreEngine(self._buffer_size)
+            server = OwnerServer(engine, listen_sock, token)
+            server.start()
+            self._engine, self._server = engine, server
+            self._family, self._address, self._token = family, address, token
+            self._role = "owner"
+            # Best-effort cleanup of the socket/advertisement on clean exit; a
+            # crash leaves them stale, which the next election self-heals.
+            atexit.register(self.close)
+        else:
+            self._family, self._address, self._token = _read_advertisement(
+                self._addr_path
+            )
+            self._role = "client"
+
+    def connect(self) -> socket.socket:
+        self.ensure()
+        assert self._family is not None and self._token is not None
+        return connect_to_owner(self._family, self._address, self._token)
+
+    def on_owner_lost(self) -> None:
+        """The owner became unreachable; re-elect (may promote us to owner)."""
+        with self._lock:
+            if self._role == "owner":
+                return
+            self._role = None
+            self._elect()
+
+    def close(self) -> None:
+        with self._lock:
+            if self._server is not None:
+                self._server.close()
+                for path in (self._sock_path, self._addr_path):
+                    try:
+                        os.unlink(path)
+                    except OSError:
+                        pass
+            self._role = None
+
+
+class _LocalSubscription(Subscription):
+    """A topic view that reads from the owner engine directly (owner role) or
+    long-polls it over a dedicated connection (client role), resuming from its
+    cursor on reconnect so no buffered message is missed.
+    """
+
+    def __init__(self, coord: _Coordinator, topic: str, start_seq: int):
+        self._coord = coord
+        self._topic = topic
+        self._cursor = start_seq
+        self._conn: Optional[socket.socket] = None
+        self._closed = threading.Event()
+
+    def close(self) -> None:
+        self._closed.set()
+        conn, self._conn = self._conn, None
+        if conn is not None:
+            try:
+                conn.close()
+            except OSError:
+                pass
+
+    def _poll_once(self) -> PollResult:
+        if self._coord.is_owner():
+            engine = self._coord.engine
+            assert engine is not None
+            return engine.poll(self._topic, self._cursor, _OWNER_POLL_TIMEOUT)
+        return self._client_poll()
+
+    def _client_poll(self) -> PollResult:
+        for attempt in range(4):
+            if self._closed.is_set():
+                return PollResult([], self._cursor, False)
+            if self._conn is None:
+                prev_token = self._coord.token
+                try:
+                    self._conn = self._coord.connect()
+                except OSError as exc:
+                    # Owner unreachable: brief retries to the same owner, then
+                    # re-elect. A changed owner means its buffer is gone -> gap.
+                    if attempt >= 2:
+                        self._coord.on_owner_lost()
+                        if self._coord.token != prev_token:
+                            raise SharedStorageGap(
+                                "shared-storage owner changed; buffered "
+                                "messages were lost"
+                            ) from exc
+                    time.sleep(0.1 * (attempt + 1))
+                    continue
+            try:
+                send_frame(
+                    self._conn,
+                    ["poll", self._topic, self._cursor, _CLIENT_POLL_TIMEOUT],
+                )
+                resp = recv_frame(self._conn)
+                if resp is EOF:
+                    raise ConnectionError("owner closed the connection")
+                status, val = resp
+                if status == "err":
+                    raise SharedStorageError(val)
+                return PollResult(*val)
+            except (OSError, EOFError, ConnectionError):
+                if self._conn is not None:
+                    try:
+                        self._conn.close()
+                    except OSError:
+                        pass
+                self._conn = None  # reconnect on the next attempt
+        return PollResult([], self._cursor, False)
+
+    def __iter__(self):
+        try:
+            while not self._closed.is_set():
+                res = self._poll_once()
+                if res.gap:
+                    raise SharedStorageGap(
+                        f"replay buffer overran on topic {self._topic!r}"
+                    )
+                yield from res.messages
+                self._cursor = res.last_seq
+        finally:
+            self.close()
+
+    def __aiter__(self):
+        return self._aiter()
+
+    async def _aiter(self):
+        loop = asyncio.get_running_loop()
+        try:
+            while not self._closed.is_set():
+                try:
+                    res = await loop.run_in_executor(None, self._poll_once)
+                except RuntimeError:
+                    # The loop/executor is shutting down (client disconnected or
+                    # the app is stopping) -- end the subscription cleanly.
+                    break
+                if res.gap:
+                    raise SharedStorageGap(
+                        f"replay buffer overran on topic {self._topic!r}"
+                    )
+                for message in res.messages:
+                    yield message
+                self._cursor = res.last_seq
+        finally:
+            self.close()
+
+
+class LocalSharedStorage(BaseSharedStorage):
+    """In-memory shared storage, elected to a single owner process per machine.
+
+    Values and published messages must be JSON-compatible. State is not durable:
+    if the owning process dies, a survivor re-elects with an empty store.
+    """
+
+    def __init__(
+        self, namespace: Optional[str] = None, buffer_size: int = DEFAULT_BUFFER
+    ):
+        self._coord = _Coordinator(namespace or _default_namespace(), buffer_size)
+        self._conn: Optional[socket.socket] = None
+        self._conn_lock = threading.Lock()
+
+    def start(self) -> None:
+        self._coord.ensure()
+
+    def close(self) -> None:
+        with self._conn_lock:
+            if self._conn is not None:
+                try:
+                    self._conn.close()
+                except OSError:
+                    pass
+                self._conn = None
+        self._coord.close()
+
+    # --- key/value + publish (short request/response) ----------------------
+    def _call(self, req):
+        last_err: Optional[Exception] = None
+        for _ in range(3):
+            if self._coord.is_owner():
+                return self._local(req)
+            try:
+                return self._remote(req)
+            except (OSError, EOFError, ConnectionError) as err:
+                last_err = err
+                self._coord.on_owner_lost()
+        raise SharedStorageError(f"shared-storage owner unreachable: {last_err}")
+
+    def _local(self, req):
+        engine = self._coord.engine
+        assert engine is not None
+        op = req[0]
+        if op == "get":
+            return engine.get(req[1], req[2])
+        if op == "set":
+            return engine.set(req[1], req[2])
+        if op == "delete":
+            return engine.delete(req[1])
+        if op == "publish":
+            return engine.publish(req[1], req[2])
+        if op == "head":
+            return engine.head_seq(req[1])
+        raise ValueError(f"unknown op {op!r}")
+
+    def _remote(self, req):
+        with self._conn_lock:
+            if self._conn is None:
+                self._conn = self._coord.connect()
+            try:
+                send_frame(self._conn, req)
+                resp = recv_frame(self._conn)
+                if resp is EOF:
+                    raise ConnectionError("owner closed the connection")
+            except (OSError, EOFError, ConnectionError):
+                try:
+                    self._conn.close()
+                except OSError:
+                    pass
+                self._conn = None
+                raise
+        status, val = resp
+        if status == "err":
+            raise SharedStorageError(val)
+        return val
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._call(["get", key, default])
+
+    def set(self, key: str, value: Any) -> None:
+        self._call(["set", key, value])
+
+    def delete(self, key: str) -> None:
+        self._call(["delete", key])
+
+    def publish(self, topic: str, message: Any) -> None:
+        self._call(["publish", topic, message])
+
+    def _head(self, topic: str) -> int:
+        return self._call(["head", topic])
+
+    def subscribe(self, topic: str, replay_from: Optional[int] = None) -> Subscription:
+        self._coord.ensure()
+        start = replay_from if replay_from is not None else self._head(topic)
+        return _LocalSubscription(self._coord, topic, start)
diff --git a/dash/_shared_storage/redis.py b/dash/_shared_storage/redis.py
new file mode 100644
index 0000000000..1702e1da66
--- /dev/null
+++ b/dash/_shared_storage/redis.py
@@ -0,0 +1,153 @@
+"""Redis-backed shared storage, for multi-process AND multi-container use.
+
+Key/value on Redis strings; ordered, replayable pub/sub on a Redis Stream per
+topic. This is the backend for horizontally-scaled deployments -- e.g. Dash
+Enterprise apps scaled across pods behind a load balancer -- where the
+single-machine ``LocalSharedStorage`` / ``DiskcacheSharedStorage`` backends
+cannot share state across containers. Redis is the single source of truth, so
+there is no owner election.
+
+Sequence numbers are assigned by an atomic server-side script (``INCR`` +
+``XADD``) so concurrent publishers stay strictly ordered; the stream is capped
+to a bounded window (``MAXLEN``), and a subscriber that falls past the trimmed
+floor gets a ``SharedStorageGap``.
+"""
+import os
+from typing import Any, Optional
+
+from ._codec import decode, encode
+from ._engine import DEFAULT_BUFFER, PollResult
+from ._polling import PollingSubscription
+from .base import BaseSharedStorage, Subscription
+
+# Redis Stream XREAD blocks server-side, so a longer cycle than the diskcache
+# sleep-poll is fine; close() latency is bounded by this.
+_POLL_TIMEOUT = 5.0
+_DEFAULT_URL = "redis://localhost:6379"
+
+# Allocate the next sequence and append atomically, so concurrent publishers
+# never produce out-of-order stream IDs. Exact MAXLEN (not '~') keeps the replay
+# window at exactly buffer_size, so the gap boundary is deterministic.
+# KEYS: seq counter, stream key. ARGV: encoded payload, maxlen.
+_PUBLISH_LUA = """
+local seq = redis.call('INCR', KEYS[1])
+redis.call('XADD', KEYS[2], 'MAXLEN', ARGV[2], seq .. '-0', 'm', ARGV[1])
+return seq
+"""
+
+
+def _require_redis():
+    try:
+        import redis  # type: ignore[import-not-found,import-untyped] # pylint: disable=import-outside-toplevel
+
+        return redis
+    except ImportError as exc:
+        raise ImportError(
+            "RedisSharedStorage requires the redis extra:\n\n"
+            '    $ pip install "dash[redis]"\n'
+        ) from exc
+
+
+def _seq_of(stream_id: Any) -> int:
+    """Integer sequence from a Redis stream id (``b"7-0"`` / ``"7-0"``)."""
+    if isinstance(stream_id, bytes):
+        stream_id = stream_id.decode()
+    return int(stream_id.split("-")[0])
+
+
+def _payload_of(fields: dict) -> bytes:
+    return fields[b"m"] if b"m" in fields else fields["m"]
+
+
+class RedisSharedStorage(BaseSharedStorage):
+    """Shared storage backed by Redis. Works across processes and containers.
+
+    Pass a ``client`` (an existing ``redis.Redis``, e.g. a Celery result
+    backend's) to reuse one connection pool, or a ``url`` (defaults to
+    ``$REDIS_URL`` then ``redis://localhost:6379``). Values and published
+    messages must be JSON-compatible. A passed client must return bytes
+    (``decode_responses=False``, the default).
+    """
+
+    def __init__(
+        self,
+        url: Optional[str] = None,
+        client: Any = None,
+        key_prefix: str = "dash:ss",
+        buffer_size: int = DEFAULT_BUFFER,
+    ):
+        redis = _require_redis()
+        if client is not None:
+            self._redis = client
+            self._owns_client = False
+        else:
+            url = url or os.environ.get("REDIS_URL") or _DEFAULT_URL
+            self._redis = redis.Redis.from_url(url)
+            self._owns_client = True
+        self._prefix = key_prefix
+        self._buffer_size = buffer_size
+        self._publish_script: Any = None
+
+    def start(self) -> None:
+        if self._publish_script is None:
+            self._publish_script = self._redis.register_script(_PUBLISH_LUA)
+
+    def close(self) -> None:
+        if self._owns_client:
+            try:
+                self._redis.close()
+            except Exception:  # pylint: disable=broad-except
+                pass
+
+    # --- key layout --------------------------------------------------------
+    def _kv(self, key: str) -> str:
+        return f"{self._prefix}:kv:{key}"
+
+    def _seq(self, topic: str) -> str:
+        return f"{self._prefix}:seq:{topic}"
+
+    def _stream(self, topic: str) -> str:
+        return f"{self._prefix}:stream:{topic}"
+
+    # --- key/value ---------------------------------------------------------
+    def get(self, key: str, default: Any = None) -> Any:
+        raw = self._redis.get(self._kv(key))
+        return default if raw is None else decode(raw)
+
+    def set(self, key: str, value: Any) -> None:
+        self._redis.set(self._kv(key), encode(value))
+
+    def delete(self, key: str) -> None:
+        self._redis.delete(self._kv(key))
+
+    # --- pub/sub -----------------------------------------------------------
+    def publish(self, topic: str, message: Any) -> None:
+        self.start()
+        self._publish_script(
+            keys=[self._seq(topic), self._stream(topic)],
+            args=[encode(message), self._buffer_size],
+        )
+
+    def _head(self, topic: str) -> int:
+        raw = self._redis.get(self._seq(topic))
+        return int(raw) if raw is not None else 0
+
+    def _poll(self, topic: str, after_seq: int, timeout: float) -> PollResult:
+        stream = self._stream(topic)
+        # Gap: the next wanted sequence sits below the trimmed floor. Checked
+        # before XREAD, which would otherwise silently resume at the floor.
+        first = self._redis.xrange(stream, count=1)
+        if first and after_seq + 1 < _seq_of(first[0][0]):
+            return PollResult([], after_seq, True)
+        entries = self._redis.xread(
+            {stream: f"{after_seq}-0"}, block=max(1, int(timeout * 1000))
+        )
+        if not entries:
+            return PollResult([], after_seq, False)
+        items = entries[0][1]
+        messages = [decode(_payload_of(fields)) for (_id, fields) in items]
+        return PollResult(messages, _seq_of(items[-1][0]), False)
+
+    def subscribe(self, topic: str, replay_from: Optional[int] = None) -> Subscription:
+        start = replay_from if replay_from is not None else self._head(topic)
+        return PollingSubscription(topic, start, self._poll, _POLL_TIMEOUT)
diff --git a/dash/dash.py b/dash/dash.py
index fd6a8d017d..511ff64c65 100644
--- a/dash/dash.py
+++ b/dash/dash.py
@@ -16,7 +16,17 @@
 import hashlib
 import base64
 from urllib.parse import urlparse
-from typing import Any, Callable, Dict, Optional, Union, Sequence, Literal, List
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Optional,
+    Type,
+    Union,
+    Sequence,
+    Literal,
+    List,
+)
 
 import traceback
 
@@ -65,6 +75,11 @@
 from . import backends
 
 from ._get_app import with_app_context, with_app_context_factory
+from ._shared_storage import (
+    BaseSharedStorage,
+    LocalSharedStorage,
+    SharedStorageError,
+)
 from ._grouping import map_grouping, grouping_len, update_args_group
 from ._obsolete import ObsoleteChecker
 from ._callback_context import callback_context
@@ -493,6 +508,9 @@ def __init__(  # pylint: disable=too-many-statements, too-many-branches
         websocket_heartbeat_interval: Optional[int] = 30000,
         websocket_batch_delay: Optional[float] = 0.005,
         websocket_max_workers: Optional[int] = 4,
+        shared_storage: Optional[
+            Union[Type[BaseSharedStorage], BaseSharedStorage]
+        ] = LocalSharedStorage,
         enable_mcp: Optional[bool] = None,
         mcp_path: Optional[str] = None,
         **obsolete,
@@ -669,6 +687,14 @@ def __init__(  # pylint: disable=too-many-statements, too-many-branches
         self._websocket_batch_delay = websocket_batch_delay
         self._websocket_max_workers = websocket_max_workers
 
+        # Shared storage (state manager + pub/sub). Started lazily on first
+        # access so it costs nothing until used and never binds in a gunicorn
+        # preload master or the Flask reloader parent -- only in the worker that
+        # actually touches it.
+        self._shared_storage_arg = shared_storage
+        self._shared_storage_instance: Optional[BaseSharedStorage] = None
+        self._shared_storage_lock = threading.Lock()
+
         self.logger = logging.getLogger(__name__)
 
         if not self.logger.handlers and add_log_handler:
@@ -929,6 +955,29 @@ def layout(self, value: Any):
             _validate.validate_layout(value, layout_value)
             self.validation_layout = layout_value
 
+    @property
+    def shared_storage(self) -> BaseSharedStorage:
+        """The app's shared storage (state manager + pub/sub), backend-agnostic.
+
+        Started on first access in the worker that uses it. Access it from a
+        callback via ``dash.ctx.shared_storage``. Raises if the app was created
+        with ``shared_storage=None``.
+        """
+        if self._shared_storage_arg is None:
+            raise SharedStorageError(
+                "Shared storage is disabled for this app; construct it with "
+                "Dash(shared_storage=...) to use dash.ctx.shared_storage."
+            )
+        if self._shared_storage_instance is None:
+            with self._shared_storage_lock:
+                if self._shared_storage_instance is None:
+                    storage = self._shared_storage_arg
+                    if isinstance(storage, type):
+                        storage = storage()
+                    storage.start()
+                    self._shared_storage_instance = storage
+        return self._shared_storage_instance
+
     def _layout_value(self):
         if self._layout_is_function:
             layout = self._layout()  # type: ignore[reportOptionalCall]
diff --git a/requirements/install.txt b/requirements/install.txt
index 176d9c8f5e..795f0c85c8 100644
--- a/requirements/install.txt
+++ b/requirements/install.txt
@@ -10,3 +10,4 @@ setuptools
 janus>=1.0.0
 pydantic>=2.10
 comm
+msgspec
diff --git a/requirements/redis.txt b/requirements/redis.txt
new file mode 100644
index 0000000000..5cc9c3dc9e
--- /dev/null
+++ b/requirements/redis.txt
@@ -0,0 +1,4 @@
+# Dependency used by RedisSharedStorage.
+# Kept in sync with the redis pin in celery.txt; in Dash 5.0 redis will be
+# removed from the celery extra and live here only.
+redis>=3.5.3,<=5.0.4
diff --git a/setup.py b/setup.py
index 3b5f7eeb20..ac7163e384 100644
--- a/setup.py
+++ b/setup.py
@@ -35,6 +35,7 @@ def read_req_file(req_type):
         "testing": read_req_file("testing"),
         "celery": read_req_file("celery"),
         "diskcache": read_req_file("diskcache"),
+        "redis": read_req_file("redis"),
         "compress": read_req_file("compress"),
         "fastapi": read_req_file("fastapi"),
         "quart": read_req_file("quart"),
diff --git a/tests/shared_storage/__init__.py b/tests/shared_storage/__init__.py
new file mode 100644
index 0000000000..4c3c33707f
--- /dev/null
+++ b/tests/shared_storage/__init__.py
@@ -0,0 +1 @@
+# Shared-storage (state manager + pub/sub) tests.
diff --git a/tests/shared_storage/test_dash_integration.py b/tests/shared_storage/test_dash_integration.py
new file mode 100644
index 0000000000..284e125993
--- /dev/null
+++ b/tests/shared_storage/test_dash_integration.py
@@ -0,0 +1,131 @@
+"""Shared storage in real Dash apps, driven through a browser with dash_duo.
+
+Each test starts a real server and clicks real buttons. The core scenario proves
+the app-facing contract: one callback writes shared state, a *different* callback
+reads it back, and the value survives across independent requests. The same
+scenario runs on each backend (Local, Diskcache, Redis-when-reachable) to prove
+the wiring end to end.
+"""
+import os
+import uuid
+
+import pytest
+
+from dash import Dash, Input, Output, ctx, html
+from dash._shared_storage import (
+    DiskcacheSharedStorage,
+    LocalSharedStorage,
+    RedisSharedStorage,
+    SharedStorageError,
+)
+
+REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
+
+
+def _redis_available():
+    try:
+        import redis  # pylint: disable=import-outside-toplevel
+
+        client = redis.Redis.from_url(REDIS_URL)
+        client.ping()
+        client.close()
+        return True
+    except Exception:  # pylint: disable=broad-except
+        return False
+
+
+def _counter_app(storage):
+    """A two-callback app: 'bump' increments a shared counter, 'read' (a separate
+    callback) shows the current shared value."""
+    app = Dash(__name__, shared_storage=storage)
+    app.layout = html.Div(
+        [
+            html.Button("bump", id="bump"),
+            html.Button("read", id="read"),
+            html.Div(id="count"),
+            html.Div(id="readout"),
+        ]
+    )
+
+    @app.callback(
+        Output("count", "children"),
+        Input("bump", "n_clicks"),
+        prevent_initial_call=True,
+    )
+    def bump(_n):
+        total = ctx.shared_storage.get("count", 0) + 1
+        ctx.shared_storage.set("count", total)
+        return str(total)
+
+    @app.callback(
+        Output("readout", "children"),
+        Input("read", "n_clicks"),
+        prevent_initial_call=True,
+    )
+    def read(_n):
+        return f"count={ctx.shared_storage.get('count', 0)}"
+
+    return app
+
+
+def _drive_counter(dash_duo):
+    """Click through the counter scenario and assert on the rendered DOM."""
+    dash_duo.find_element("#bump").click()
+    dash_duo.wait_for_text_to_equal("#count", "1")
+    dash_duo.find_element("#bump").click()
+    dash_duo.wait_for_text_to_equal("#count", "2")
+
+    # A different callback reads the value the bump callback wrote.
+    dash_duo.find_element("#read").click()
+    dash_duo.wait_for_text_to_equal("#readout", "count=2")
+
+    assert dash_duo.get_logs() == []
+
+
+def test_counter_shared_across_callbacks_local(dash_duo):
+    dash_duo.start_server(
+        _counter_app(LocalSharedStorage(namespace=f"duo-{uuid.uuid4().hex[:12]}"))
+    )
+    _drive_counter(dash_duo)
+
+
+def test_counter_shared_across_callbacks_diskcache(dash_duo, tmp_path):
+    dash_duo.start_server(
+        _counter_app(DiskcacheSharedStorage(directory=str(tmp_path / "ss")))
+    )
+    _drive_counter(dash_duo)
+
+
+def test_counter_shared_across_callbacks_redis(dash_duo):
+    if not _redis_available():
+        pytest.skip("no Redis reachable at REDIS_URL")
+    dash_duo.start_server(
+        _counter_app(
+            RedisSharedStorage(
+                url=REDIS_URL, key_prefix=f"dash:duo:{uuid.uuid4().hex[:12]}"
+            )
+        )
+    )
+    _drive_counter(dash_duo)
+
+
+def test_disabled_shared_storage_errors_the_callback(dash_duo):
+    """With shared_storage=None, a callback touching ctx.shared_storage fails;
+    the app surfaces a callback error rather than updating the output."""
+    app = Dash(__name__, shared_storage=None)
+    app.layout = html.Div([html.Button("go", id="go"), html.Div(id="out")])
+
+    @app.callback(
+        Output("out", "children"), Input("go", "n_clicks"), prevent_initial_call=True
+    )
+    def cb(_n):
+        return ctx.shared_storage.get("x")
+
+    dash_duo.start_server(app)
+    dash_duo.find_element("#go").click()
+    # The output never fills in, and the server logged the SharedStorageError.
+    dash_duo.wait_for_text_to_equal("#out", "")
+    assert dash_duo.get_logs()  # an error was logged
+
+    with pytest.raises(SharedStorageError):
+        app.shared_storage  # noqa: B018
diff --git a/tests/shared_storage/test_diskcache_backend.py b/tests/shared_storage/test_diskcache_backend.py
new file mode 100644
index 0000000000..7ec83eafac
--- /dev/null
+++ b/tests/shared_storage/test_diskcache_backend.py
@@ -0,0 +1,139 @@
+"""Tests for DiskcacheSharedStorage (KV + sequenced pub/sub on a disk cache).
+
+Covers the contract semantics (future-only subscribe, replay, gap) in-process
+and cross-process visibility over a shared cache directory. Requires the
+diskcache extra, which the shared-storage CI job installs.
+"""
+import multiprocessing as mp
+import threading
+import time
+
+import pytest
+
+from dash._shared_storage import DiskcacheSharedStorage, SharedStorageGap
+
+CTX = mp.get_context("spawn")
+
+
+@pytest.fixture
+def store(tmp_path):
+    s = DiskcacheSharedStorage(directory=str(tmp_path / "cache"))
+    try:
+        yield s
+    finally:
+        s.close()
+
+
+def _drain(sub, n):
+    out = []
+    for msg in sub:
+        out.append(msg)
+        if len(out) == n:
+            break
+    return out
+
+
+def test_kv_get_set_delete(store):
+    assert store.get("missing") is None
+    assert store.get("missing", 42) == 42
+    store.set("a", {"x": 1})
+    assert store.get("a") == {"x": 1}
+    store.delete("a")
+    assert store.get("a") is None
+    store.delete("a")  # idempotent
+
+
+def test_fresh_subscriber_only_sees_future_messages(store):
+    store.publish("t", "old")
+    sub = store.subscribe("t")  # cursor at current head
+    received = []
+    th = threading.Thread(target=lambda: received.extend(_drain(sub, 2)))
+    th.start()
+    time.sleep(0.2)
+    store.publish("t", "new1")
+    store.publish("t", "new2")
+    th.join(timeout=5)
+    sub.close()
+    assert received == ["new1", "new2"]
+
+
+def test_replay_from_cursor(store):
+    store.publish("t", "m1")
+    store.publish("t", "m2")
+    store.publish("t", "m3")
+    sub = store.subscribe("t", replay_from=1)  # saw up to seq 1
+    assert _drain(sub, 2) == ["m2", "m3"]
+    sub.close()
+
+
+def test_gap_when_buffer_overruns(tmp_path):
+    store = DiskcacheSharedStorage(directory=str(tmp_path / "c"), buffer_size=2)
+    for i in range(5):
+        store.publish("t", f"m{i}")  # seqs 1..5; only 4,5 retained
+    sub = store.subscribe("t", replay_from=1)  # wants seq 2, evicted
+    with pytest.raises(SharedStorageGap):
+        next(iter(sub))
+    sub.close()
+    store.close()
+
+
+def test_no_gap_at_buffer_edge(tmp_path):
+    store = DiskcacheSharedStorage(directory=str(tmp_path / "c"), buffer_size=2)
+    for i in range(4):
+        store.publish("t", f"m{i}")  # seqs 1..4; 3,4 retained
+    sub = store.subscribe("t", replay_from=2)  # wants seq 3, still held
+    assert _drain(sub, 2) == ["m2", "m3"]
+    sub.close()
+    store.close()
+
+
+# --- cross-process (shared cache directory) -------------------------------
+
+
+def _child_set(directory, key, value):
+    s = DiskcacheSharedStorage(directory=directory)
+    s.set(key, value)
+    s.close()
+
+
+def _child_publish(directory, topic, messages, start_ev):
+    s = DiskcacheSharedStorage(directory=directory)
+    start_ev.wait(timeout=10)
+    for m in messages:
+        s.publish(topic, m)
+    s.close()
+
+
+def test_kv_visible_across_processes(tmp_path):
+    directory = str(tmp_path / "shared")
+    store = DiskcacheSharedStorage(directory=directory)
+    p = CTX.Process(target=_child_set, args=(directory, "shared", {"n": 42}))
+    p.start()
+    p.join(timeout=10)
+    assert store.get("shared") == {"n": 42}
+    store.close()
+
+
+def test_pubsub_across_processes(tmp_path):
+    directory = str(tmp_path / "shared")
+    store = DiskcacheSharedStorage(directory=directory)
+    sub = store.subscribe("topic")
+
+    start_ev = CTX.Event()
+    p = CTX.Process(
+        target=_child_publish,
+        args=(directory, "topic", ["m0", "m1", "m2"], start_ev),
+    )
+    p.start()
+
+    received = []
+    th = threading.Thread(target=lambda: received.extend(_drain(sub, 3)))
+    th.start()
+    time.sleep(0.3)  # ensure the subscriber is polling before publishing
+    start_ev.set()
+
+    th.join(timeout=10)
+    p.join(timeout=10)
+    sub.close()
+    store.close()
+    assert received == ["m0", "m1", "m2"]
diff --git a/tests/shared_storage/test_engine.py b/tests/shared_storage/test_engine.py
new file mode 100644
index 0000000000..b1f849d4d0
--- /dev/null
+++ b/tests/shared_storage/test_engine.py
@@ -0,0 +1,119 @@
+"""Unit tests for the in-memory StoreEngine (KV + sequenced pub/sub)."""
+import threading
+import time
+
+from dash._shared_storage._engine import StoreEngine
+
+
+def test_kv_get_set_delete():
+    e = StoreEngine()
+    assert e.get("missing") is None
+    assert e.get("missing", 42) == 42
+    e.set("a", {"x": 1})
+    assert e.get("a") == {"x": 1}
+    e.delete("a")
+    assert e.get("a") is None
+    e.delete("a")  # idempotent
+
+
+def test_publish_assigns_monotonic_seq():
+    e = StoreEngine()
+    assert e.head_seq("t") == 0
+    assert e.publish("t", "a") == 1
+    assert e.publish("t", "b") == 2
+    assert e.head_seq("t") == 2
+
+
+def test_fresh_subscriber_only_sees_future_messages():
+    e = StoreEngine()
+    e.publish("t", "old")
+    cursor = e.head_seq("t")  # subscribe "now"
+    e.publish("t", "new1")
+    e.publish("t", "new2")
+    res = e.poll("t", cursor, timeout=1)
+    assert res.messages == ["new1", "new2"]
+    assert res.last_seq == 3  # "old" took seq 1, so new2 is seq 3
+    assert res.gap is False
+
+
+def test_replay_from_cursor_after_reconnect():
+    e = StoreEngine()
+    e.publish("t", "m1")
+    e.publish("t", "m2")
+    e.publish("t", "m3")
+    # A consumer that saw up to seq 1 reconnects and replays 2 and 3.
+    res = e.poll("t", 1, timeout=1)
+    assert res.messages == ["m2", "m3"]
+    assert res.last_seq == 3
+    assert res.gap is False
+
+
+def test_gap_when_buffer_overruns():
+    e = StoreEngine(buffer_size=2)
+    for i in range(5):
+        e.publish("t", f"m{i}")  # seqs 1..5, buffer holds only seqs 4,5
+    # A consumer stuck at seq 1 wanted seq 2, which was evicted -> gap.
+    res = e.poll("t", 1, timeout=1)
+    assert res.gap is True
+    assert res.messages == []
+
+
+def test_no_gap_at_buffer_edge():
+    e = StoreEngine(buffer_size=2)
+    for i in range(4):
+        e.publish("t", f"m{i}")  # seqs 1..4, buffer holds 3,4
+    # Consumer at seq 2 wants seq 3, which is still buffered -> no gap.
+    res = e.poll("t", 2, timeout=1)
+    assert res.gap is False
+    assert res.messages == ["m2", "m3"]
+
+
+def test_poll_times_out_empty_when_no_messages():
+    e = StoreEngine()
+    start = time.monotonic()
+    res = e.poll("t", 0, timeout=0.2)
+    assert res.messages == []
+    assert res.gap is False
+    assert time.monotonic() - start >= 0.2
+
+
+def test_poll_wakes_on_publish_from_another_thread():
+    e = StoreEngine()
+    received = []
+
+    def consumer():
+        res = e.poll("t", 0, timeout=2)
+        received.extend(res.messages)
+
+    th = threading.Thread(target=consumer)
+    th.start()
+    time.sleep(0.1)  # ensure the poll is waiting
+    e.publish("t", "live")
+    th.join(timeout=2)
+    assert received == ["live"]
+
+
+def test_close_unblocks_waiting_pollers():
+    e = StoreEngine()
+    done = threading.Event()
+
+    def consumer():
+        e.poll("t", 0, timeout=10)
+        done.set()
+
+    th = threading.Thread(target=consumer)
+    th.start()
+    time.sleep(0.1)
+    e.close()
+    assert done.wait(timeout=2)
+
+
+def test_multiple_subscribers_each_get_every_message():
+    e = StoreEngine()
+    cursor = e.head_seq("t")
+    e.publish("t", "a")
+    e.publish("t", "b")
+    a = e.poll("t", cursor, timeout=1)
+    b = e.poll("t", cursor, timeout=1)
+    assert a.messages == ["a", "b"]
+    assert b.messages == ["a", "b"]  # independent cursors, both see all
diff --git a/tests/shared_storage/test_local_cross_process.py b/tests/shared_storage/test_local_cross_process.py
new file mode 100644
index 0000000000..a45fa2745e
--- /dev/null
+++ b/tests/shared_storage/test_local_cross_process.py
@@ -0,0 +1,185 @@
+"""Cross-process tests for LocalSharedStorage.
+
+An owner runs in a spawned child process; the pytest process attaches as a
+client (the child wins election first). Exercises KV visibility, pub/sub, and
+client reconnect-with-replay across a forced socket drop -- all over the real
+socket transport, not the in-process fast path.
+"""
+import multiprocessing as mp
+import threading
+import time
+import uuid
+
+import pytest
+
+from dash._shared_storage import LocalSharedStorage
+
+CTX = mp.get_context("spawn")
+
+
+def _owner_main(namespace, cmd_q, done_q, ready_ev):
+    """A controllable owner: wins election, then runs commands on demand."""
+    store = LocalSharedStorage(namespace=namespace)
+    store.start()
+    if not store._coord.is_owner():  # pragma: no cover - defensive
+        done_q.put(("error", "child did not win election"))
+        return
+    ready_ev.set()
+    while True:
+        cmd = cmd_q.get()
+        if cmd is None:
+            break
+        op, args = cmd
+        if op == "set":
+            store.set(*args)
+        elif op == "publish":
+            store.publish(*args)
+        done_q.put(("done", op))
+    store.close()
+
+
+class _Owner:
+    """Test helper: spawn a controllable owner and drive it synchronously."""
+
+    def __init__(self, namespace):
+        self.cmd_q = CTX.Queue()
+        self.done_q = CTX.Queue()
+        self.ready = CTX.Event()
+        self.proc = CTX.Process(
+            target=_owner_main,
+            args=(namespace, self.cmd_q, self.done_q, self.ready),
+            daemon=True,
+        )
+
+    def start(self):
+        self.proc.start()
+        assert self.ready.wait(timeout=10), "owner failed to start"
+
+    def do(self, op, *args):
+        self.cmd_q.put((op, args))
+        kind, _ = self.done_q.get(timeout=10)
+        assert kind == "done"
+
+    def stop(self):
+        self.cmd_q.put(None)
+        self.proc.join(timeout=10)
+
+
+@pytest.fixture
+def owner():
+    ns = f"xproc-{uuid.uuid4().hex[:12]}"
+    o = _Owner(ns)
+    o.start()
+    try:
+        yield ns, o
+    finally:
+        o.stop()
+
+
+def test_kv_visible_across_processes(owner):
+    ns, o = owner
+    client = LocalSharedStorage(namespace=ns)
+    client.start()
+    assert not client._coord.is_owner()  # child owns; we are the client
+
+    o.do("set", "shared", {"n": 42})
+    assert client.get("shared") == {"n": 42}
+    assert client.get("nope", "fallback") == "fallback"
+    client.close()
+
+
+def test_pubsub_across_processes(owner):
+    ns, o = owner
+    client = LocalSharedStorage(namespace=ns)
+    client.start()
+
+    sub = client.subscribe("topic")  # cursor at current head
+    received = []
+
+    def consume():
+        for msg in sub:
+            received.append(msg)
+            if len(received) == 3:
+                break
+
+    th = threading.Thread(target=consume)
+    th.start()
+    time.sleep(0.3)  # ensure the long-poll is established
+
+    for i in range(3):
+        o.do("publish", "topic", f"m{i}")
+
+    th.join(timeout=10)
+    sub.close()
+    client.close()
+    assert received == ["m0", "m1", "m2"]
+
+
+def test_client_reconnect_replays_missed_messages(owner):
+    ns, o = owner
+    client = LocalSharedStorage(namespace=ns)
+    client.start()
+
+    sub = client.subscribe("stream")
+    received = []
+    ready = threading.Event()
+
+    def consume():
+        ready.set()
+        for msg in sub:
+            received.append(msg)
+            if len(received) == 6:
+                break
+
+    th = threading.Thread(target=consume)
+    th.start()
+    ready.wait()
+    time.sleep(0.3)
+
+    o.do("publish", "stream", "a")
+    o.do("publish", "stream", "b")
+
+    # Wait until the first two land, then forcibly drop the client's socket.
+    _wait_until(lambda: len(received) >= 2, timeout=5)
+    conn = sub._conn
+    if conn is not None:
+        conn.close()  # simulate a proxy idle-timeout / network blip
+
+    # Messages published during/after the drop must still arrive (buffer replay).
+    for msg in ("c", "d", "e", "f"):
+        o.do("publish", "stream", msg)
+
+    th.join(timeout=10)
+    sub.close()
+    client.close()
+    assert received == ["a", "b", "c", "d", "e", "f"]
+
+
+def test_reelection_after_owner_killed(owner):
+    ns, o = owner
+    client = LocalSharedStorage(namespace=ns)
+    client.start()
+    assert not client._coord.is_owner()
+
+    o.do("set", "before", "value")
+    assert client.get("before") == "value"
+
+    # Kill the owner without a clean shutdown (leaves a stale socket).
+    o.proc.terminate()
+    o.proc.join(timeout=10)
+
+    # The survivor re-elects: it becomes the new (cold) owner and keeps serving.
+    assert client.get("before") is None  # cold store, prior state gone
+    client.set("after", "fresh")
+    assert client.get("after") == "fresh"
+    assert client._coord.is_owner()  # we are the new owner
+    client.close()
+
+
+def _wait_until(pred, timeout):
+    end = time.monotonic() + timeout
+    while time.monotonic() < end:
+        if pred():
+            return
+        time.sleep(0.02)
+    raise AssertionError("condition not met in time")
diff --git a/tests/shared_storage/test_redis_backend.py b/tests/shared_storage/test_redis_backend.py
new file mode 100644
index 0000000000..3d99c130fa
--- /dev/null
+++ b/tests/shared_storage/test_redis_backend.py
@@ -0,0 +1,137 @@
+"""Tests for RedisSharedStorage (KV + Redis Streams pub/sub).
+
+Requires a reachable Redis (``$REDIS_URL`` or ``redis://localhost:6379``); the
+tests skip when none is available. The CI shared-storage job provides one. Each
+test uses a unique key prefix so a shared Redis stays isolated.
+"""
+import os
+import threading
+import time
+import uuid
+
+import pytest
+
+redis = pytest.importorskip("redis")
+
+# pylint: disable=wrong-import-position
+from dash._shared_storage import (  # noqa: E402
+    RedisSharedStorage,
+    SharedStorageGap,
+)
+
+REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
+
+
+def _redis_available():
+    try:
+        client = redis.Redis.from_url(REDIS_URL)
+        client.ping()
+        client.close()
+        return True
+    except Exception:  # pylint: disable=broad-except
+        return False
+
+
+pytestmark = pytest.mark.skipif(
+    not _redis_available(), reason="no Redis reachable at REDIS_URL"
+)
+
+
+@pytest.fixture
+def store():
+    prefix = f"dash:sstest:{uuid.uuid4().hex[:12]}"
+    s = RedisSharedStorage(url=REDIS_URL, key_prefix=prefix)
+    s.start()
+    try:
+        yield s
+    finally:
+        s.close()
+
+
+def _drain(sub, n):
+    out = []
+    for msg in sub:
+        out.append(msg)
+        if len(out) == n:
+            break
+    return out
+
+
+def test_kv_get_set_delete(store):
+    assert store.get("missing") is None
+    assert store.get("missing", 42) == 42
+    store.set("a", {"x": 1})
+    assert store.get("a") == {"x": 1}
+    store.delete("a")
+    assert store.get("a") is None
+    store.delete("a")  # idempotent
+
+
+def test_fresh_subscriber_only_sees_future_messages(store):
+    store.publish("t", "old")
+    sub = store.subscribe("t")  # cursor at current head
+    received = []
+    th = threading.Thread(target=lambda: received.extend(_drain(sub, 2)))
+    th.start()
+    time.sleep(0.3)
+    store.publish("t", "new1")
+    store.publish("t", "new2")
+    th.join(timeout=5)
+    sub.close()
+    assert received == ["new1", "new2"]
+
+
+def test_replay_from_cursor(store):
+    store.publish("t", "m1")
+    store.publish("t", "m2")
+    store.publish("t", "m3")
+    sub = store.subscribe("t", replay_from=1)  # saw up to seq 1
+    assert _drain(sub, 2) == ["m2", "m3"]
+    sub.close()
+
+
+def test_gap_when_buffer_overruns():
+    prefix = f"dash:sstest:{uuid.uuid4().hex[:12]}"
+    store = RedisSharedStorage(url=REDIS_URL, key_prefix=prefix, buffer_size=2)
+    store.start()
+    for i in range(5):
+        store.publish("t", f"m{i}")  # seqs 1..5; stream trimmed to 4,5
+    sub = store.subscribe("t", replay_from=1)  # wants seq 2, trimmed away
+    with pytest.raises(SharedStorageGap):
+        next(iter(sub))
+    sub.close()
+    store.close()
+
+
+def test_no_gap_at_buffer_edge():
+    prefix = f"dash:sstest:{uuid.uuid4().hex[:12]}"
+    store = RedisSharedStorage(url=REDIS_URL, key_prefix=prefix, buffer_size=2)
+    store.start()
+    for i in range(4):
+        store.publish("t", f"m{i}")  # seqs 1..4; stream holds 3,4
+    sub = store.subscribe("t", replay_from=2)  # wants seq 3, still held
+    assert _drain(sub, 2) == ["m2", "m3"]
+    sub.close()
+    store.close()
+
+
+def test_two_instances_share_state(store):
+    """A second client (separate connection pool) sees the first's writes and
+    published messages -- the multi-worker / multi-pod case."""
+    other = RedisSharedStorage(url=REDIS_URL, key_prefix=store._prefix)
+    other.start()
+
+    store.set("shared", {"n": 42})
+    assert other.get("shared") == {"n": 42}
+
+    sub = other.subscribe("topic")
+    received = []
+    th = threading.Thread(target=lambda: received.extend(_drain(sub, 3)))
+    th.start()
+    time.sleep(0.3)
+    for i in range(3):
+        store.publish("topic", f"m{i}")
+    th.join(timeout=5)
+    sub.close()
+    other.close()
+    assert received == ["m0", "m1", "m2"]