diff --git a/python/reflex_xy/__init__.py b/python/reflex_xy/__init__.py
index 0a554722..7e184303 100644
--- a/python/reflex_xy/__init__.py
+++ b/python/reflex_xy/__init__.py
@@ -54,6 +54,9 @@ def index() -> rx.Component:
     "set_view": ".app",
     "setup": ".app",
     "chart": ".component",
+    "AsyncDataVar": ".data_vars",
+    "DataVar": ".data_vars",
+    "data": ".data_vars",
     "DataHandle": ".handles",
     "FigureHandle": ".handles",
     "CanonicalRowIdGroup": ".events",
@@ -78,10 +81,12 @@ def index() -> rx.Component:
 
 __all__ = [
     "XY_NAMESPACE",
+    "AsyncDataVar",
     "AsyncFigureVar",
     "CanonicalRowIdGroup",
     "DataBounds",
     "DataHandle",
+    "DataVar",
     "FigureHandle",
     "FigureRegistry",
     "FigureVar",
@@ -98,6 +103,7 @@ def index() -> rx.Component:
     "append",
     "chart",
     "clear_selection",
+    "data",
     "figure",
     "inline",
     "register",
@@ -231,6 +237,7 @@ def release(token: "str | FigureHandle") -> None:
 if TYPE_CHECKING:
     from .app import XYPlugin, append, clear_selection, reset_view, select, set_view, setup
     from .component import chart
+    from .data_vars import AsyncDataVar, DataVar, data
     from .events import (
         CanonicalRowIdGroup,
         DataBounds,
diff --git a/python/reflex_xy/data_vars.py b/python/reflex_xy/data_vars.py
new file mode 100644
index 00000000..0ba1573b
--- /dev/null
+++ b/python/reflex_xy/data_vars.py
@@ -0,0 +1,226 @@
+"""`@reflex_xy.data`: a computed var that *is* the dataset registration.
+
+The data-plane sibling of `@reflex_xy.figure` (vars.py) for the data-bound
+component tier: the state method returns **columns only** — a mapping of
+column name to array-likes — so there is no chart API inside it to get
+wrong. Evaluating the var publishes the columns into the per-process
+registry under a deterministic token (`xyd1|<client>|<state>|<var>`) and
+its value is a tiny typed :class:`~reflex_xy.handles.DataHandle`. Reflex's
+dependency tracking watches the method body; a state change republishes the
+columns, and the registry rebuilds + broadcasts every mounted chart plan
+bound to them.
+
+The method's return annotation is the compile-time schema channel (fact
+R7): annotate a ``TypedDict`` and the class-level var carries
+``DataHandle[Schema]``, which the chart factories read column names from —
+without executing any user code. A plain ``dict[str, ...]`` annotation
+degrades gracefully to first-execution validation.
+
+Like figure builders, data methods must be pure functions of their state
+instance: the token is the rebuild recipe (state_bridge.py re-runs the
+method when a fresh worker needs the columns back), and purity is what
+makes the column set a rebuildable cache instead of precious process state.
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Callable, Mapping
+from typing import Any, Optional, get_type_hints, is_typeddict, overload
+
+from reflex_base.vars.base import AsyncComputedVar, ComputedVar
+
+from .handles import DataHandle
+from .registry import registry
+from .tokens import BUILDER_ATTR, build_data_token
+from .vars import _builder_target
+
+__all__ = ["AsyncDataVar", "DataVar", "data", "validate_columns"]
+
+
+class DataVar(ComputedVar):
+    """ComputedVar whose value is a DataHandle (sync data method)."""
+
+    def _deps(self, objclass: Any, obj: Any = None) -> dict[str, set[str]]:
+        return ComputedVar._deps(self, objclass, obj=_builder_target(self, obj))
+
+
+class AsyncDataVar(AsyncComputedVar):
+    """AsyncComputedVar whose value is a DataHandle (async data method)."""
+
+    def _deps(self, objclass: Any, obj: Any = None) -> dict[str, set[str]]:
+        return AsyncComputedVar._deps(self, objclass, obj=_builder_target(self, obj))
+
+
+def validate_columns(columns: Any, *, source: str) -> dict[str, Any]:
+    """The only checks that need real data: a mapping of named, equal-length
+    array-like columns. Everything structural was validated at compile by the
+    plan's zero-row probe; dtype/shape details stay with figure compilation."""
+    if not isinstance(columns, Mapping):
+        raise TypeError(
+            f"{source} must return a mapping of column name -> values "
+            f"(e.g. a TypedDict of arrays), got {type(columns).__name__}"
+        )
+    validated: dict[str, Any] = {}
+    lengths: dict[str, int] = {}
+    for key, values in columns.items():
+        if not isinstance(key, str):
+            raise TypeError(f"{source} column names must be strings, got {key!r}")
+        if isinstance(values, (str, bytes, Mapping)):
+            raise TypeError(
+                f"{source} column {key!r} must be an array-like of values, "
+                f"got {type(values).__name__}"
+            )
+        try:
+            lengths[key] = len(values)
+        except TypeError as exc:
+            raise TypeError(
+                f"{source} column {key!r} must be an array-like with a length, "
+                f"got {type(values).__name__}"
+            ) from exc
+        validated[key] = values
+    if len(set(lengths.values())) > 1:
+        detail = ", ".join(f"{key}={length}" for key, length in lengths.items())
+        raise ValueError(f"{source} columns must share one length, got {detail}")
+    return validated
+
+
+def _mint_token(state: Any, var_name: str) -> Optional[str]:
+    client_token = state.router.session.client_token
+    if not client_token:
+        return None
+    return build_data_token(client_token, type(state).get_full_name(), var_name)
+
+
+def _publish(token: str, columns: Any, *, source: str) -> DataHandle:
+    if columns is None:
+        registry.release_columns(token)
+        return DataHandle("")
+    registry.publish_columns(token, validate_columns(columns, source=source))
+    return DataHandle(token)
+
+
+def _source_label(method: Callable[..., Any]) -> str:
+    qualname = getattr(method, "__qualname__", None) or getattr(method, "__name__", "data method")
+    return qualname.rsplit(".<locals>.", 1)[-1]
+
+
+def _adopt_identity(fget: Any, method: Callable[..., Any], name: str) -> None:
+    fget.__name__ = name
+    fget.__qualname__ = getattr(method, "__qualname__", name)
+    fget.__module__ = getattr(method, "__module__", fget.__module__)
+    fget.__doc__ = method.__doc__
+    setattr(fget, BUILDER_ATTR, method)
+
+
+def _make_fget(method: Callable[[Any], Any]) -> Callable[[Any], DataHandle]:
+    name = _fn_name(method)
+    source = _source_label(method)
+
+    def fget(self: Any) -> DataHandle:
+        token = _mint_token(self, name)
+        if token is None:
+            return DataHandle("")
+        return _publish(token, method(self), source=source)
+
+    _adopt_identity(fget, method, name)
+    return fget
+
+
+def _make_async_fget(method: Callable[[Any], Any]) -> Callable[[Any], Any]:
+    name = _fn_name(method)
+    source = _source_label(method)
+
+    async def fget(self: Any) -> DataHandle:
+        token = _mint_token(self, name)
+        if token is None:
+            return DataHandle("")
+        return _publish(token, await method(self), source=source)
+
+    _adopt_identity(fget, method, name)
+    return fget
+
+
+def _fn_name(fn: Callable[..., Any]) -> str:
+    name = getattr(fn, "__name__", "")
+    if not name:
+        msg = f"@reflex_xy.data methods must be named functions, got {fn!r}"
+        raise TypeError(msg)
+    return name
+
+
+def _return_type(method: Callable[..., Any]) -> Any:
+    """``DataHandle[Schema]`` when the return annotation is a TypedDict —
+    the schema survives as the class-level var's ``_var_type`` (R7) and is
+    how the factories compile-check column names. Anything else (plain
+    dicts, missing or unresolvable annotations) degrades to ``DataHandle``:
+    columns are then validated on first execution instead."""
+    try:
+        hints = get_type_hints(method)
+    except Exception:  # noqa: BLE001 - annotations may reference unimportable names
+        return DataHandle
+    annotation = hints.get("return")
+    if annotation is not None and is_typeddict(annotation):
+        return DataHandle[annotation]
+    return DataHandle
+
+
+@overload
+def data(method: Callable[[Any], Any]) -> "DataVar | AsyncDataVar": ...
+
+
+@overload
+def data(
+    method: None = None, **var_kwargs: Any
+) -> Callable[[Callable[[Any], Any]], "DataVar | AsyncDataVar"]: ...
+
+
+def data(
+    method: Optional[Callable[[Any], Any]] = None, **var_kwargs: Any
+) -> "DataVar | AsyncDataVar | Callable[[Callable[[Any], Any]], DataVar | AsyncDataVar]":
+    """Declare a chart dataset on a Reflex state class.
+
+    Usage::
+
+        class CloudData(TypedDict):
+            x: np.ndarray
+            y: np.ndarray
+            mag: np.ndarray
+
+        class Dash(rx.State):
+            points: int = 200_000
+
+            @reflex_xy.data
+            def cloud(self) -> CloudData:
+                rng = np.random.default_rng(7)
+                x = rng.normal(size=self.points)
+                return {"x": x, "y": x * 0.6, "mag": np.abs(x)}
+
+        # in the page:
+        #   reflex_xy.scatter_chart(data=Dash.cloud, x="x", y="y", color="mag")
+
+    The method must return a mapping of column name -> equal-length
+    array-likes, or ``None`` for "no data right now" (which releases the
+    registered columns and yields the empty handle). ``async def`` methods
+    become ``AsyncDataVar``s (same dispatch rule as ``rx.var``); keyword
+    arguments pass through to reflex's computed var (``deps=``,
+    ``interval=``, ...).
+    """
+
+    def _decorate(fn: Callable[[Any], Any]) -> "DataVar | AsyncDataVar":
+        if _fn_name(fn).startswith("_"):
+            # Same rule as figure vars: the handle must sync to the client,
+            # and backend (underscore) vars never do.
+            msg = (
+                "@reflex_xy.data vars must not start with '_' (the handle must sync to the client)"
+            )
+            raise ValueError(msg)
+        var_kwargs.setdefault("cache", True)
+        return_type = _return_type(fn)
+        if inspect.iscoroutinefunction(fn):
+            return AsyncDataVar(fget=_make_async_fget(fn), return_type=return_type, **var_kwargs)
+        return DataVar(fget=_make_fget(fn), return_type=return_type, **var_kwargs)
+
+    if method is None:
+        return _decorate
+    return _decorate(method)
diff --git a/python/reflex_xy/plan.py b/python/reflex_xy/plan.py
new file mode 100644
index 00000000..28f54628
--- /dev/null
+++ b/python/reflex_xy/plan.py
@@ -0,0 +1,244 @@
+"""Chart plans: validated, data-free chart structure for the data-bound tier.
+
+A plan is the server-side half of the composite figure identity
+``xyp1|<digest>|<data token>`` (spec/design/reflex-integration.md): the xy
+node tree with **string channels only**, compiled once at page evaluation.
+
+- **Build** (factory call = page evaluation = Reflex compile): construct the
+  real xy tree, bind a zero-row placeholder column for every referenced
+  channel name, and call ``.figure()`` once — the full mark/config
+  validation gate (facts X1/X2, pinned in tests/test_validation_timing.py)
+  runs in milliseconds with no real data. The probe figure is discarded;
+  what is kept is the digest, the recorded column names, and (for live
+  charts) the probe's Tailwind class inventory.
+- **Serialize**: nodes → canonical JSON (sorted keys, ``plan_version``) →
+  sha256 prefix = ``digest``. The digest is a *content address*: every
+  worker that evaluates the page derives the same digest and holds the plan
+  in this process-local map — and backend-only workers, which skip the
+  frontend compile, get the same evaluation from ``setup(app)``'s startup
+  lifespan (app.py `_ensure_page_plans`). A lookup miss is hot-reload
+  drift and answers ``err {resync}`` naming the digest.
+- **Bind** (serve time): columns + plan → a **fresh** ``Chart`` (never
+  reuse — ``Chart.figure()`` memoizes, X3) → ``Figure``. Column mismatches
+  raise :class:`PlanBindError` naming both sides.
+
+Plans hold no data and no session identity; they are pure functions of page
+source. Everything session-shaped lives in the data token half.
+"""
+
+from __future__ import annotations
+
+import copy
+import dataclasses
+import hashlib
+import json
+from collections.abc import Mapping
+from typing import Any, Optional
+
+import numpy as np
+
+from xy.components import Chart, Component, Mark
+
+__all__ = [
+    "PLAN_VERSION",
+    "ChartPlan",
+    "PlanBindError",
+    "PlanError",
+    "PlanMissError",
+    "build_plan",
+    "plan_of",
+    "register_plan",
+]
+
+PLAN_VERSION = 1
+_DIGEST_CHARS = 20  # sha256 hex prefix; content address for a process-local map
+
+#: Mark kinds whose figure-compile validators require at least one finite
+#: value (they aggregate: quantiles, bins, meshes). The zero-row probe
+#: cannot compile them, so they are excluded from the plan tier — the
+#: Phase 3 decision recorded in reflex-component-api-implementation.md.
+_NEEDS_DATA_MARKS = frozenset({"box", "violin", "hexbin", "contour", "heatmap", "stairs", "ecdf"})
+
+
+class PlanError(ValueError):
+    """A chart plan could not be built, resolved, or bound."""
+
+
+class PlanMissError(PlanError):
+    """No plan registered under a digest (hot-reload drift — client resyncs)."""
+
+    def __init__(self, digest: str) -> None:
+        super().__init__(
+            f"unknown chart plan {digest!r} on this worker (stale page after a "
+            "hot reload?); re-subscribe against the recompiled page"
+        )
+        self.digest = digest
+
+
+class PlanBindError(PlanError):
+    """The data var's columns do not satisfy the plan's bindings."""
+
+
+class _ProbeTable(Mapping):
+    """Zero-row placeholder table that records every column it resolves.
+
+    ``Chart.figure()`` resolves string channels through ``data[name]``
+    (the exact production code path), so the recorded names are *derived*
+    from the real resolution logic — the plan's column list can never drift
+    from what binding will actually look up.
+    """
+
+    def __init__(self) -> None:
+        self.seen: list[str] = []
+
+    def __getitem__(self, key: str) -> np.ndarray:
+        if key not in self.seen:
+            self.seen.append(key)
+        return np.empty(0, dtype=np.float64)
+
+    def __iter__(self):  # pragma: no cover - Mapping protocol completeness
+        return iter(self.seen)
+
+    def __len__(self) -> int:  # pragma: no cover - Mapping protocol completeness
+        return len(self.seen)
+
+
+def _plain(value: Any, context: str) -> Any:
+    """Canonical JSON-able copy of one plan node field (fail closed)."""
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        node: dict[str, Any] = {"~node": type(value).__name__}
+        for field in dataclasses.fields(value):
+            node[field.name] = _plain(getattr(value, field.name), f"{context}.{field.name}")
+        return node
+    if isinstance(value, Mapping):
+        return {str(key): _plain(item, f"{context}.{key}") for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_plain(item, context) for item in value]
+    if isinstance(value, np.generic):
+        return value.item()
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    raise PlanError(
+        f"{context} holds a {type(value).__name__}, which cannot be part of a "
+        "data-bound chart plan. Plans are data-free structure: bind columns "
+        "by name (strings) and keep arrays in the @reflex_xy.data method; "
+        "components (e.g. legend/tooltip render=) belong to the escape "
+        "hatch (@reflex_xy.figure) or the static tier."
+    )
+
+
+@dataclasses.dataclass(frozen=True)
+class ChartPlan:
+    """One validated, data-free chart structure, addressed by content."""
+
+    kind: str
+    children: tuple[Component, ...]
+    chart_props: dict[str, Any]
+    columns: tuple[str, ...]  # channel names the probe resolved, in order
+    tailwind_classes: str  # probe figure's DOM class inventory (live-tier scan)
+    digest: str
+
+    def chart(self, data: Any) -> Chart:
+        """A fresh ``Chart`` over ``data`` (X3: never reuse a compiled one)."""
+        return Chart(self.kind, self.children, data=data, **self.chart_props)
+
+    def bind(self, columns: Mapping[str, Any], *, source: str = "the data var") -> Chart:
+        """Bind real columns; missing bindings name both sides."""
+        missing = [name for name in self.columns if name not in columns]
+        if missing:
+            bound = ", ".join(repr(name) for name in missing)
+            produced = ", ".join(sorted(str(key) for key in columns)) or "no columns"
+            noun = "columns" if len(missing) > 1 else "column"
+            raise PlanBindError(f"plan binds {noun} {bound}; {source} produced {{{produced}}}")
+        return self.chart(columns)
+
+
+def build_plan(
+    kind: str, children: tuple[Component, ...], chart_props: dict[str, Any]
+) -> ChartPlan:
+    """Compile + validate a plan and register it in this worker's map."""
+    for child in children:
+        if isinstance(child, Mark) and child.data is not None:
+            raise PlanError(
+                "per-mark data= is not supported in data-bound charts; bind one "
+                "chart-level data source (this is tracked as deferred work in "
+                "spec/design/reflex-component-api-implementation.md)"
+            )
+    # The plan must be immutable once addressed: hash a deep snapshot and
+    # register *that* snapshot, so mutating a reused mark node or a props
+    # dict after the factory call can never change binding behavior behind
+    # an unchanged digest/columns record.
+    children = copy.deepcopy(children)
+    chart_props = copy.deepcopy(chart_props)
+    serialized = {
+        "plan_version": PLAN_VERSION,
+        "kind": kind,
+        "chart": {key: _plain(value, f"{kind}() {key}") for key, value in chart_props.items()},
+        "children": [_plain(child, f"{kind}() child {i}") for i, child in enumerate(children)],
+    }
+    canonical = json.dumps(serialized, sort_keys=True, separators=(",", ":"))
+    digest = hashlib.sha256(canonical.encode()).hexdigest()[:_DIGEST_CHARS]
+
+    # The compile-time validation gate: bind zero-row placeholders for every
+    # string channel and compile once. Errors surface here — at page
+    # evaluation — with the ordinary xy messages.
+    probe = _ProbeTable()
+    try:
+        probe_figure = Chart(kind, children, data=probe, **chart_props).figure()
+    except ValueError as exc:
+        needy = sorted(
+            {
+                child.kind
+                for child in children
+                if isinstance(child, Mark) and child.kind in _NEEDS_DATA_MARKS
+            }
+        )
+        if needy:
+            raise PlanError(
+                f"{', '.join(needy)} marks aggregate their values, so their "
+                "validators need at least one row — the zero-row plan probe "
+                "cannot compile them. Data-bound charts exclude these kinds "
+                "(recorded in reflex-component-api-implementation.md, Phase 3 "
+                "decision); build the chart with @reflex_xy.figure, or pass "
+                "a concrete xy Chart to reflex_xy.chart() for the static tier."
+            ) from exc
+        raise
+    tailwind_classes = " ".join(probe_figure.dom_class_strings())
+
+    plan = ChartPlan(
+        kind=kind,
+        children=children,
+        chart_props=chart_props,
+        columns=tuple(probe.seen),
+        tailwind_classes=tailwind_classes,
+        digest=digest,
+    )
+    return register_plan(plan)
+
+
+#: Process-local {digest: plan}, populated wherever page bodies evaluate:
+#: the frontend compile, and — for backend-only workers that skip it — the
+#: setup(app) startup lifespan (app.py `_ensure_page_plans`). Entries are
+#: tiny (node dataclasses with string channels) and bounded by page code.
+_PLANS: dict[str, ChartPlan] = {}
+
+
+def register_plan(plan: ChartPlan) -> ChartPlan:
+    """Idempotently register a plan under its digest; returns the canonical one."""
+    return _PLANS.setdefault(plan.digest, plan)
+
+
+def plan_of(digest: str) -> Optional[ChartPlan]:
+    return _PLANS.get(digest)
+
+
+def require_plan(digest: str) -> ChartPlan:
+    plan = _PLANS.get(digest)
+    if plan is None:
+        raise PlanMissError(digest)
+    return plan
+
+
+def reset_plans_for_tests() -> None:
+    """Forget every registered plan (test isolation only)."""
+    _PLANS.clear()
diff --git a/python/reflex_xy/registry.py b/python/reflex_xy/registry.py
index 8986923d..6ea57d8b 100644
--- a/python/reflex_xy/registry.py
+++ b/python/reflex_xy/registry.py
@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import threading
 import time
 import uuid
@@ -28,7 +29,9 @@
 if TYPE_CHECKING:
     from xy._figure import Figure
 
-__all__ = ["FigureEntry", "FigureRegistry", "registry"]
+__all__ = ["ColumnEntry", "FigureEntry", "FigureRegistry", "registry"]
+
+_logger = logging.getLogger("reflex_xy")
 
 # Idle figures are swept after this long without a subscribe/message/publish.
 # Deterministic (state-backed) figures rebuild transparently on the next
@@ -94,11 +97,38 @@ def touch(self) -> None:
         self.last_access = time.monotonic()
 
 
+@dataclass
+class ColumnEntry:
+    """One registered column set (the data half of the data-bound tier).
+
+    Column entries are pure rebuildable caches of Reflex state — the
+    `@reflex_xy.data` method is the recipe — so they carry none of the
+    figure entry's kernel machinery: no locks (a republish replaces the
+    whole immutable generation), no pins, always sweepable.
+    """
+
+    columns: dict[str, Any]
+    token: str
+    version: int = 1
+    last_access: float = field(default_factory=time.monotonic)
+
+    def touch(self) -> None:
+        self.last_access = time.monotonic()
+
+
 class FigureRegistry:
     """token -> FigureEntry map with versioning and TTL sweep."""
 
     def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
         self._entries: dict[str, FigureEntry] = {}
+        # Data-bound tier: column sets published by @reflex_xy.data vars,
+        # and the data-token -> {plan digest} index that lets a column
+        # republish rebuild + broadcast every mounted dependent figure. The
+        # index is bounded by mounted plans: entries are added when a
+        # composite figure binds and dropped when a republish finds neither
+        # a cached figure nor live subscribers under the composite token.
+        self._column_entries: dict[str, ColumnEntry] = {}
+        self._digests_by_data_token: dict[str, set[str]] = {}
         # A mounted client remains in its socket room when a TTL sweep evicts
         # a rebuildable figure. Retain only the evicted scalar version while
         # at least one such client is subscribed, so a state-driven rebuild on
@@ -134,6 +164,11 @@ def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
         # async callback(token, message, buffers, version) -> None for append
         # and view-state pushes — the message-shaped data-plane seam.
         self._on_push: Optional[PushHook] = None
+        # async callback(token, error, resync) -> None: room-wide err frames
+        # for server-side failures with no request to answer (a data
+        # republish whose plan bind fails must not leave stale pixels
+        # silently frozen).
+        self._on_error: Optional[Callable[[str, str, bool], Awaitable[None]]] = None
 
     # -- wiring ------------------------------------------------------------
 
@@ -149,6 +184,28 @@ def on_push(
     ) -> None:
         self._on_push = callback
 
+    def on_error(self, callback: Callable[[str, str, bool], Awaitable[None]]) -> None:
+        self._on_error = callback
+
+    def _schedule_error(self, token: str, error: str, *, resync: bool) -> None:
+        """Fan an out-of-band failure to a figure room from any thread."""
+        callback = self._on_error
+        loop = self._loop
+        if callback is None or loop is None:
+            return
+
+        async def _run() -> None:
+            await callback(token, error, resync)
+
+        try:
+            running = asyncio.get_running_loop()
+        except RuntimeError:
+            running = None
+        if running is loop:
+            loop.create_task(_run())
+        else:
+            asyncio.run_coroutine_threadsafe(_run(), loop)
+
     def _enqueue_push(
         self,
         entry: FigureEntry,
@@ -277,38 +334,7 @@ def publish(
         which is the signal that subscribers need a new payload.
         """
         with self._mutex:
-            entry = self._entries.get(token)
-            # A cache-miss rebuild inserts its entry before room fan-out. A
-            # canonical same-object publish re-authorizes that provisional
-            # generation without bumping its version, but must still own a
-            # normal broadcast in case the rebuild-owned fan-out fails.
-            reauthorized = entry is not None and bool(self._active_rebuild_guards.get(token))
-            # This is a canonical dependency/application publish. Any builder
-            # that began from the preceding absence must not overwrite it,
-            # even if this entry is released again before that builder ends.
-            self._invalidate_rebuild_guards_locked(token)
-            if entry is None:
-                version = self._evicted_versions.pop(token, 0) + 1
-                entry = FigureEntry(figure=figure, token=token, version=version, pinned=pinned)
-                self._entries[token] = entry
-                changed = True
-            else:
-                changed = entry.figure is not figure
-                if changed:
-                    # A replacement is a new immutable generation. In-flight
-                    # handlers keep a self-consistent old figure/version pair
-                    # instead of observing the old figure with a newly bumped
-                    # version. Their results are discarded by ``is_current``.
-                    entry = FigureEntry(
-                        figure=figure,
-                        token=token,
-                        version=entry.version + 1,
-                        pinned=entry.pinned or pinned,
-                    )
-                    self._entries[token] = entry
-                else:
-                    entry.pinned = entry.pinned or pinned
-                    entry.touch()
+            entry, changed, reauthorized = self._publish_locked(token, figure, pinned)
         if broadcast and (changed or reauthorized):
             # Re-publishing the identical object means nothing moved; a new
             # figure object is normally the signal subscribers need a fresh
@@ -318,6 +344,48 @@ def publish(
             self.schedule_broadcast(token)
         return entry
 
+    def _publish_locked(
+        self, token: str, figure: "Figure", pinned: bool
+    ) -> tuple[FigureEntry, bool, bool]:
+        """The mutation half of ``publish``; the registry mutex must be held.
+
+        Returns ``(entry, changed, reauthorized)`` so callers can schedule the
+        broadcast after releasing the mutex.
+        """
+        entry = self._entries.get(token)
+        # A cache-miss rebuild inserts its entry before room fan-out. A
+        # canonical same-object publish re-authorizes that provisional
+        # generation without bumping its version, but must still own a
+        # normal broadcast in case the rebuild-owned fan-out fails.
+        reauthorized = entry is not None and bool(self._active_rebuild_guards.get(token))
+        # This is a canonical dependency/application publish. Any builder
+        # that began from the preceding absence must not overwrite it,
+        # even if this entry is released again before that builder ends.
+        self._invalidate_rebuild_guards_locked(token)
+        if entry is None:
+            version = self._evicted_versions.pop(token, 0) + 1
+            entry = FigureEntry(figure=figure, token=token, version=version, pinned=pinned)
+            self._entries[token] = entry
+            changed = True
+        else:
+            changed = entry.figure is not figure
+            if changed:
+                # A replacement is a new immutable generation. In-flight
+                # handlers keep a self-consistent old figure/version pair
+                # instead of observing the old figure with a newly bumped
+                # version. Their results are discarded by ``is_current``.
+                entry = FigureEntry(
+                    figure=figure,
+                    token=token,
+                    version=entry.version + 1,
+                    pinned=entry.pinned or pinned,
+                )
+                self._entries[token] = entry
+            else:
+                entry.pinned = entry.pinned or pinned
+                entry.touch()
+        return entry, changed, reauthorized
+
     def begin_rebuild(self, token: str) -> tuple[Optional[FigureEntry], Optional[object]]:
         """Atomically return the current entry or guard this missing generation."""
         with self._mutex:
@@ -478,6 +546,114 @@ def __len__(self) -> int:
         with self._mutex:
             return len(self._entries)
 
+    # -- data-bound tier: column entries + plan index ------------------------
+
+    def publish_columns(self, token: str, columns: dict[str, Any]) -> ColumnEntry:
+        """Insert or replace a column set, then rebuild every mounted plan
+        bound to it (fresh figures under the composite tokens, broadcast by
+        the ordinary publish fan-out). Entries are immutable generations: a
+        republish replaces the whole entry and bumps its version."""
+        with self._mutex:
+            previous = self._column_entries.get(token)
+            version = 1 if previous is None else previous.version + 1
+            entry = ColumnEntry(columns=columns, token=token, version=version)
+            self._column_entries[token] = entry
+            digests = sorted(self._digests_by_data_token.get(token, ()))
+        for digest in digests:
+            self._rebuild_dependent(token, digest, entry)
+        return entry
+
+    def get_columns(self, token: str) -> Optional[ColumnEntry]:
+        with self._mutex:
+            entry = self._column_entries.get(token)
+            if entry is not None:
+                entry.touch()
+            return entry
+
+    def release_columns(self, token: str) -> None:
+        """Drop a column set and every dependent composite figure entry.
+
+        The "no data right now" path (`@reflex_xy.data` returning None): the
+        empty handle unmounts subscribed charts client-side, and dropping
+        the dependent figure caches here keeps a later re-mount rebuilding
+        from current state instead of serving pre-release columns.
+        """
+        from .tokens import build_plan_token
+
+        with self._mutex:
+            self._column_entries.pop(token, None)
+            digests = self._digests_by_data_token.pop(token, set())
+        for digest in sorted(digests):
+            self.release(build_plan_token(digest, token))
+
+    def bind_plan(self, data_token: str, digest: str) -> None:
+        """Record that a mounted plan binds this data token (idempotent)."""
+        with self._mutex:
+            self._digests_by_data_token.setdefault(data_token, set()).add(digest)
+
+    def _rebuild_dependent(self, data_token: str, digest: str, column_entry: ColumnEntry) -> None:
+        """Re-bind one dependent plan against freshly published columns.
+
+        ``column_entry`` is the generation this rebuild serves. Concurrent
+        republishes rebuild concurrently and may complete out of order; every
+        outcome below (publish, failure release + err frame) is gated on the
+        generation still being current — atomically with the registry mutation
+        — so an older generation that finishes late can never overwrite a
+        newer one's figure or report a stale failure.
+        """
+        from .plan import plan_of
+        from .tokens import build_plan_token, parse_token
+
+        composite = build_plan_token(digest, data_token)
+        with self._mutex:
+            if self._column_entries.get(data_token) is not column_entry:
+                return  # a newer generation owns this rebuild now
+            mounted = composite in self._entries or bool(
+                self._rebuildable_subscribers.get(composite)
+            )
+            if not mounted:
+                # Nothing serves or watches this plan anymore: forget the
+                # binding (this is what keeps the index bounded by mounts);
+                # a later subscribe rebuilds from scratch and re-indexes.
+                digests = self._digests_by_data_token.get(data_token)
+                if digests is not None:
+                    digests.discard(digest)
+                    if not digests:
+                        self._digests_by_data_token.pop(data_token, None)
+                return
+        parsed = parse_token(data_token)
+        source = (
+            f"{parsed.state_full_name}.{parsed.var_name}" if parsed is not None else "the data var"
+        )
+        plan = plan_of(digest)
+        try:
+            if plan is None:
+                from .plan import PlanMissError
+
+                raise PlanMissError(digest)
+            figure = plan.bind(column_entry.columns, source=source).figure()
+        except Exception as exc:  # noqa: BLE001 - user data meets page structure here
+            # Fail loud on both sides: the server log carries the bind error,
+            # and subscribers get an err frame instead of silently frozen
+            # pixels (their bounded resync retries against current state) —
+            # unless a newer generation superseded this one meanwhile, in
+            # which case the failure is obsolete and it owns the outcome.
+            with self._mutex:
+                if self._column_entries.get(data_token) is not column_entry:
+                    return
+                self._invalidate_rebuild_guards_locked(composite)
+                removed = self._entries.pop(composite, None)
+                self._retain_removed_version_locked(composite, removed)
+            _logger.warning("republish of %s failed: %s", composite, exc)
+            self._schedule_error(composite, str(exc), resync=True)
+            return
+        with self._mutex:
+            if self._column_entries.get(data_token) is not column_entry:
+                return  # a newer generation's figure must win; drop this one
+            _, changed, reauthorized = self._publish_locked(composite, figure, False)
+        if changed or reauthorized:
+            self.schedule_broadcast(composite)
+
     # -- version bump + fan-out ---------------------------------------------
 
     def bump(self, token: str, *, expected: Optional[FigureEntry] = None) -> Optional[FigureEntry]:
@@ -703,7 +879,12 @@ def clear_selection(self, token: str) -> None:
     # -- TTL sweep -----------------------------------------------------------
 
     def sweep(self, *, now: Optional[float] = None) -> list[str]:
-        """Drop unpinned entries idle past the TTL; returns dropped tokens."""
+        """Drop unpinned entries idle past the TTL; returns dropped tokens.
+
+        Column entries sweep under the same TTL: they are state-rebuildable
+        caches exactly like state-token figures (a later subscribe re-runs
+        the data method), so the sweep bounds column memory, not correctness.
+        """
         now = time.monotonic() if now is None else now
         dropped: list[str] = []
         with self._mutex:
@@ -717,6 +898,10 @@ def sweep(self, *, now: Optional[float] = None) -> list[str]:
                     self._retain_removed_version_locked(token, entry)
                     del self._entries[token]
                     dropped.append(token)
+            for token, column_entry in list(self._column_entries.items()):
+                if now - column_entry.last_access > self._ttl:
+                    del self._column_entries[token]
+                    dropped.append(token)
         return dropped
 
     async def sweep_forever(self) -> None:
@@ -734,6 +919,8 @@ async def sweep_forever(self) -> None:
 def reset_registry_for_tests() -> FigureRegistry:
     """Reset the process registry in place (test isolation only)."""
     registry._entries.clear()
+    registry._column_entries.clear()
+    registry._digests_by_data_token.clear()
     registry._evicted_versions.clear()
     registry._rebuildable_subscribers.clear()
     registry._rebuildable_tokens_by_sid.clear()
@@ -742,6 +929,7 @@ def reset_registry_for_tests() -> FigureRegistry:
     registry._loop = None
     registry._on_publish = None
     registry._on_push = None
+    registry._on_error = None
     registry._ttl = DEFAULT_TTL_SECONDS
     return registry
 
diff --git a/python/reflex_xy/tokens.py b/python/reflex_xy/tokens.py
index 049dad5f..9aed0308 100644
--- a/python/reflex_xy/tokens.py
+++ b/python/reflex_xy/tokens.py
@@ -1,22 +1,31 @@
-"""Figure tokens: the only chart-related value that lives in Reflex state.
-
-Two token families, one namespace:
-
-- **State tokens** (`xyv1|<client_token>|<state_full_name>|<var_name>`) are
-  minted by the `@reflex_xy.figure` computed var. They are *deterministic*:
-  any backend worker holding the same Reflex state can re-derive the figure
-  from the token alone, which is what makes reconnects and multi-worker
-  deployments work without a central figure store (the token IS the recipe;
-  Reflex state is the pantry).
-- **Opaque tokens** (`xyfig-<uuid>`) come from imperative
-  `reflex_xy.register(...)`. They cannot be rebuilt elsewhere — dev-tier by
-  design, documented in spec/design/reflex-integration.md.
+"""Chart tokens: the only chart-related values that live in Reflex state.
+
+Three token families, one namespace:
+
+- **Figure state tokens** (`xyv1|<client_token>|<state_full_name>|<var_name>`)
+  are minted by the `@reflex_xy.figure` computed var. They are
+  *deterministic*: any backend worker holding the same Reflex state can
+  re-derive the figure from the token alone, which is what makes reconnects
+  and multi-worker deployments work without a central figure store (the
+  token IS the recipe; Reflex state is the pantry).
+- **Data state tokens** (`xyd1|<client_token>|<state_full_name>|<var_name>`)
+  are minted by the `@reflex_xy.data` computed var under the same grammar
+  and rebuild contract, naming a registered *column set* instead of a
+  figure. They never subscribe alone: the wrapper composes them with a plan
+  digest into a **plan token** (`xyp1|<digest>|<data token>`), the composite
+  figure identity of the data-bound tier — plan digest for the structure,
+  data token for the columns; both halves independently recoverable.
+- **Opaque tokens** (`xyfig-<uuid>` / `xyin-<digest>`) come from imperative
+  `reflex_xy.register(...)` / `inline(...)`. They cannot be rebuilt
+  elsewhere — dev/fixed tiers by design, documented in
+  spec/design/reflex-integration.md.
 
 Tokens are visible to their own client (they ride through state deltas), so
 they must not carry anything the client doesn't already know: the client
-token is the browser tab's own session id, and state/var names already
-appear in every state delta. Cross-client use is refused by the namespace's
-affinity check, not by token secrecy.
+token is the browser tab's own session id, state/var names already appear
+in every state delta, and a plan digest is a content address of page code
+the client was compiled from. Cross-client use is refused by the
+namespace's affinity check, not by token secrecy.
 """
 
 from __future__ import annotations
@@ -27,23 +36,32 @@
 from typing import Any, Optional
 
 __all__ = [
+    "ParsedPlanToken",
     "ParsedToken",
+    "build_data_token",
+    "build_plan_token",
     "build_state_token",
     "builder_of",
+    "parse_plan_token",
     "parse_token",
 ]
 
-_PREFIX = "xyv1"
+_FIGURE_PREFIX = "xyv1"
+_DATA_PREFIX = "xyd1"
+_PLAN_PREFIX = "xyp1"
 _SEP = "|"
 # Client tokens are UUID-ish; state full names and var names are dotted
 # Python identifiers. Nothing here may contain the separator.
-_TOKEN_RE = re.compile(
-    r"^xyv1\|(?P<client>[A-Za-z0-9_-]{8,64})"
+_STATE_TOKEN_BODY = (
+    r"(?P<client>[A-Za-z0-9_-]{8,64})"
     r"\|(?P<state>[A-Za-z0-9_.]{1,512})"
     r"\|(?P<var>[A-Za-z_][A-Za-z0-9_]{0,255})$"
 )
+_TOKEN_RE = re.compile(r"^(?P<prefix>xyv1|xyd1)\|" + _STATE_TOKEN_BODY)
+# Plan digests are lowercase sha256 hex prefixes (plan.py).
+_PLAN_RE = re.compile(r"^xyp1\|(?P<digest>[0-9a-f]{8,64})\|(?P<data>xyd1\|.+)$")
 
-#: Attribute stashed on a figure var's fget carrying the user's builder.
+#: Attribute stashed on a figure/data var's fget carrying the user's builder.
 #: It lives on the *function* (not the ComputedVar) so it survives reflex's
 #: `_replace` copies, which re-instantiate the var but thread fget through.
 BUILDER_ATTR = "__xy_builder__"
@@ -51,23 +69,55 @@
 
 @dataclass(frozen=True)
 class ParsedToken:
+    """A parsed state token; ``kind`` is ``"figure"`` (xyv1) or ``"data"`` (xyd1)."""
+
     client_token: str
     state_full_name: str
     var_name: str
+    kind: str = "figure"
 
 
-def build_state_token(client_token: str, state_full_name: str, var_name: str) -> str:
-    token = _SEP.join((_PREFIX, client_token, state_full_name, var_name))
-    if parse_token(token) is None:
+@dataclass(frozen=True)
+class ParsedPlanToken:
+    """A parsed composite plan token: plan digest + the embedded data token."""
+
+    digest: str
+    data: ParsedToken
+    data_token: str  # the verbatim xyd1|… substring (registry key for columns)
+
+
+def _build_state_token(prefix: str, client_token: str, state_full_name: str, var_name: str) -> str:
+    token = _SEP.join((prefix, client_token, state_full_name, var_name))
+    if _TOKEN_RE.match(token) is None:
         # Defensive: a state or client token that defeats the grammar would
         # otherwise mint a token the namespace can never resolve.
-        msg = f"cannot build a valid figure token from {client_token!r}/{state_full_name!r}/{var_name!r}"
+        msg = f"cannot build a valid {prefix} token from {client_token!r}/{state_full_name!r}/{var_name!r}"
+        raise ValueError(msg)
+    return token
+
+
+def build_state_token(client_token: str, state_full_name: str, var_name: str) -> str:
+    """Mint a figure token (`xyv1|…`)."""
+    return _build_state_token(_FIGURE_PREFIX, client_token, state_full_name, var_name)
+
+
+def build_data_token(client_token: str, state_full_name: str, var_name: str) -> str:
+    """Mint a data token (`xyd1|…`)."""
+    return _build_state_token(_DATA_PREFIX, client_token, state_full_name, var_name)
+
+
+def build_plan_token(digest: str, data_token: str) -> str:
+    """Compose the data-bound tier's figure identity (`xyp1|<digest>|<data>`)."""
+    token = _SEP.join((_PLAN_PREFIX, digest, data_token))
+    if parse_plan_token(token) is None:
+        msg = f"cannot build a valid plan token from {digest!r}/{data_token!r}"
         raise ValueError(msg)
     return token
 
 
 def parse_token(token: str) -> Optional[ParsedToken]:
-    """Parse a state token; None for opaque/foreign strings (fail closed)."""
+    """Parse a state token (figure or data); None for opaque/foreign strings
+    (fail closed). Composite plan tokens parse via :func:`parse_plan_token`."""
     if not isinstance(token, str):
         return None
     match = _TOKEN_RE.match(token)
@@ -77,11 +127,26 @@ def parse_token(token: str) -> Optional[ParsedToken]:
         client_token=match["client"],
         state_full_name=match["state"],
         var_name=match["var"],
+        kind="figure" if match["prefix"] == _FIGURE_PREFIX else "data",
     )
 
 
+def parse_plan_token(token: str) -> Optional[ParsedPlanToken]:
+    """Parse a composite plan token; None for anything else (fail closed)."""
+    if not isinstance(token, str):
+        return None
+    match = _PLAN_RE.match(token)
+    if match is None:
+        return None
+    data = parse_token(match["data"])
+    if data is None or data.kind != "data":
+        return None
+    return ParsedPlanToken(digest=match["digest"], data=data, data_token=match["data"])
+
+
 def builder_of(state_cls: Any, var_name: str) -> Optional[Callable[[Any], Any]]:
-    """Find the figure builder a `@reflex_xy.figure` var attached to a state class."""
+    """Find the builder a `@reflex_xy.figure` / `@reflex_xy.data` var
+    attached to a state class."""
     computed = getattr(state_cls, "computed_vars", None)
     var = computed.get(var_name) if isinstance(computed, dict) else None
     fget = getattr(var, "_fget", None)
diff --git a/python/reflex_xy/vars.py b/python/reflex_xy/vars.py
index 53f7190c..4898956f 100644
--- a/python/reflex_xy/vars.py
+++ b/python/reflex_xy/vars.py
@@ -46,8 +46,9 @@
 
 
 def _builder_target(var: Any, obj: Any) -> Any:
-    """Point dependency tracking at the *builder*, not the token wrapper:
-    reflex should track what the chart reads, and the wrapper fget reads
+    """Point dependency tracking at the wrapped method — the figure builder
+    here, the data method in data_vars.py — not the token wrapper: reflex
+    should track what the method body reads, and the wrapper fget reads
     nothing but the router."""
     if obj is not None:
         return obj
diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md
index 23b4a9ec..4f97b6cf 100644
--- a/spec/design/reflex-integration.md
+++ b/spec/design/reflex-integration.md
@@ -425,6 +425,100 @@ generations are leased; the sweep skips them until the mutation and version
 bump finish. Rapid re-publishes coalesce: an un-started broadcast absorbs newer
 publishes and always ships the latest payload.
 
+### 3.6 The data-bound tier: `@reflex_xy.data`, plans, composite tokens
+
+The primary component API (adopted design: Option 6 of
+[`reflex-component-api-options.md`](reflex-component-api-options.md); work
+plan in [`reflex-component-api-implementation.md`](reflex-component-api-implementation.md))
+splits the figure var's job in two: **structure is declared in the page**,
+**state supplies only columns**.
+
+```python
+class CloudData(TypedDict):
+    x: np.ndarray; y: np.ndarray; mag: np.ndarray
+
+class Dash(rx.State):
+    points: int = 200_000
+
+    @reflex_xy.data                     # columns only — no chart API inside
+    def cloud(self) -> CloudData: ...
+
+def index():
+    return reflex_xy.scatter_chart(     # flat form; reflex_xy.chart(*nodes)
+        data=Dash.cloud,                # is the composed multi-mark form
+        x="x", y="y", color="mag", colormap="viridis",
+        height="460px", on_select_end=Dash.select,
+    )
+```
+
+**`@reflex_xy.data`** (`data_vars.py` — the module is named `data_vars`
+because a `data.py` submodule would shadow the `reflex_xy.data` export) is
+the exact sibling of `@reflex_xy.figure`: a computed var whose value is a
+typed `DataHandle` wrapping `xyd1|<client>|<state>|<var>` (same grammar,
+charset, purity contract, pre-session short-circuit, underscore refusal,
+async dispatch, and `None`-releases semantics as figure vars). Evaluating
+it validates the returned mapping — string keys, array-likes, one shared
+length; the only checks that need real data — and publishes the **columns**
+into the registry. The method's return annotation is the compile-time
+schema channel (fact R7): a `TypedDict` parametrizes the handle
+(`DataHandle[CloudData]`), and the factories read the column names from the
+class-level var without executing anything; a plain `dict` annotation
+degrades to first-execution validation.
+
+**Plans** (`plan.py`). A chart factory call at page evaluation compiles its
+xy nodes (string channels only) into a `ChartPlan`: the real tree is built,
+zero-row placeholder columns are bound for every referenced channel through
+the production resolution path (a recording table — the column list cannot
+drift from what binding will look up), and `.figure()` runs once — the full
+mark/config validation gate (X1/X2) at compile time, in milliseconds, with
+no data ingestion (constraint 2). The canonical JSON of the tree
+(`plan_version: 1`) is content-addressed into a sha256-prefix `digest` and
+registered in a process-local `{digest: plan}` map. Binding is the reverse:
+columns + plan → a **fresh** `Chart` (never reused — X3) → `.figure()`.
+Column-mismatch errors name both sides (*"plan binds column 'mag';
+Dash.cloud produced {x, y}"*). Plans refuse concrete arrays, per-mark
+`data=`, and `render=` components — data-free structure only. The probe
+figure also yields `dom_class_strings()`, so **live data-bound charts get
+automatic Tailwind discovery** (previously live sources needed the manual
+inventory). The factories that build plans, and the errors they catch at
+`reflex run`, are §5.
+
+**Column entries.** Published columns are registry entries in their own
+right, keyed by the data token: pure rebuildable caches of Reflex state
+(the `@reflex_xy.data` method is the recipe), so they carry none of the
+figure entry's kernel machinery — no locks (a republish replaces the whole
+immutable generation and bumps its version), no pins, always sweepable
+under the same TTL. The registry also keeps the `data token → {digests}`
+index that lets a column republish rebuild every mounted dependent plan,
+and an error seam (`on_error`) for room-wide failures that answer no
+request; §4 covers the fan-out those two enable.
+
+**Republish ordering.** Dependent rebuilds run outside the registry mutex
+(they execute user-scale figure builds), so two republishes of one data
+token may finish in reverse order. Every rebuild therefore carries the
+`ColumnEntry` generation it serves, and each of its outcomes — the figure
+publish, and the failure path's release + `err {resync}` frame — is gated
+on that generation still being current, *atomically with the registry
+mutation*. An older generation that finishes late is dropped whole:
+subscribers can never observe a newer generation's pixels replaced by an
+older one's, and a superseded failure raises no stale error. Pinned by
+`test_data_var.py::test_stale_column_generation_cannot_overwrite_a_newer_publish`.
+
+**Plan immutability.** `build_plan` hashes — and registers — a deep
+snapshot of the node tree and chart props, not the caller's objects.
+Page code that reuses and later mutates a mark node (or the props dict)
+cannot change binding behavior behind an unchanged digest/columns record;
+the canonical representation is the source of truth. Pinned by
+`test_plan.py::test_plan_is_a_snapshot_immune_to_later_node_mutation`.
+
+**Plan format stability.** `plan_version: 1` is part of the canonical
+serialization and a golden digest is pinned in
+`tests/reflex_adapter/test_plan.py`. Digests are content addresses, not a
+migration surface: after a format (or grammar) change, old subscribers'
+digests simply miss and resync against the recompiled page — the golden
+exists to catch *accidental* churn, and an intentional change bumps
+`PLAN_VERSION` and re-records it.
+
 ## 4. Updates and streaming
 
 - **State-driven rebuild** (filter changed): the figure var recomputes,
@@ -718,11 +812,15 @@ demo app models.
 ```
 python/reflex_xy/
   registry.py                token -> FigureEntry(figure, version, lock); TTL;
-                             publish/push fan-out seams; append
-  tokens.py                  xyv1 token grammar; builder discovery on vars
+                             publish/push/error fan-out seams; append; column
+                             entries + data-token -> digest index (§3.6)
+  tokens.py                  xyv1/xyd1/xyp1 token grammar; builder discovery
   handles.py                 FigureHandle / DataHandle[S] (+ serializers):
                              the typed values chart state vars carry
   vars.py                    @reflex_xy.figure (FigureVar: builder-tracked deps)
+  data_vars.py               @reflex_xy.data (DataVar: columns in, handle out)
+  plan.py                    ChartPlan: zero-row probe, canonical digest,
+                             process-local plan map, bind (§3.6)
   state_bridge.py            token -> state_manager -> builder rebuild hook
   namespace.py               XYNamespace: sub/unsub/msg, payload/msg/err,
                              affinity, rebuild-on-miss, binary attachments
@@ -746,9 +844,9 @@ examples/reflex/  (repo root) Reflex showcase: figure-var drilldown with
                              whose category toggles re-bin kernel-side, §34)
 examples/fastapi/ (repo root) the same charts + a live 100M drilldown served
                              from a plain FastAPI app (no committed HTML)
-tests/reflex_adapter/        token/registry/var/bridge/payload-asset units,
-                             component compile, framework contract pins
-                             (R1/R7/R8), and a real-websocket
+tests/reflex_adapter/        token/registry/var/data-var/plan/bridge/
+                             payload-asset units, component compile, framework
+                             contract pins (R1/R7/R8), and a real-websocket
                              integration suite (uvicorn + socketio client)
                              covering payload/pick/select/affinity/rebuild/
                              publish-broadcast/append/unsub
diff --git a/tests/reflex_adapter/test_data_var.py b/tests/reflex_adapter/test_data_var.py
new file mode 100644
index 00000000..20698d4e
--- /dev/null
+++ b/tests/reflex_adapter/test_data_var.py
@@ -0,0 +1,254 @@
+"""The @reflex_xy.data computed var: columns in, typed handle out."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import TypedDict, get_args
+
+import numpy as np
+import pytest
+import reflex as rx
+from reflex_base.vars.base import AsyncComputedVar
+
+import reflex_xy
+import xy
+from reflex_xy.data_vars import AsyncDataVar, DataVar, validate_columns
+from reflex_xy.handles import DataHandle
+from reflex_xy.plan import build_plan
+from reflex_xy.tokens import build_plan_token, builder_of, parse_token
+
+from .conftest import make_router_data
+
+
+class DataSchema(TypedDict):
+    x: np.ndarray
+    y: np.ndarray
+    mag: np.ndarray
+
+
+class DataDemo(rx.State):
+    n: int = 16
+    _scale: float = 2.0
+
+    @reflex_xy.data
+    def cloud(self) -> DataSchema:
+        xs = np.linspace(0.0, 1.0, self.n)
+        return {"x": xs, "y": xs * self._scale, "mag": np.abs(xs)}
+
+    @reflex_xy.data
+    def maybe(self):
+        if self.n < 0:
+            return None
+        return {"x": [1.0], "y": [2.0]}
+
+    @reflex_xy.data
+    async def remote(self) -> DataSchema:
+        await asyncio.sleep(0)
+        xs = np.linspace(0.0, 1.0, self.n)
+        return {"x": xs, "y": xs, "mag": xs}
+
+
+def hydrated_substate(client_token: str) -> DataDemo:
+    root = rx.State(_reflex_internal_init=True)
+    root.router = make_router_data(client_token)
+    return root.get_substate(tuple(DataDemo.get_full_name().split("."))[1:])
+
+
+def test_dispatch_and_schema_channel():
+    """R7: the TypedDict return annotation parametrizes the var's value type;
+    async methods dispatch to AsyncDataVar exactly like rx.var."""
+    assert isinstance(DataDemo.computed_vars["cloud"], DataVar)
+    assert DataDemo.cloud._var_type == DataHandle[DataSchema]
+    assert get_args(DataDemo.cloud._var_type) == (DataSchema,)
+    # untyped methods degrade to the plain handle (columns checked at runtime)
+    assert DataDemo.maybe._var_type is DataHandle
+    assert isinstance(DataDemo.computed_vars["remote"], AsyncDataVar)
+    assert isinstance(DataDemo.computed_vars["remote"], AsyncComputedVar)
+    assert DataDemo.remote._var_type == DataHandle[DataSchema]
+
+
+def test_deps_track_the_data_method_not_the_wrapper():
+    deps = DataDemo.computed_vars["cloud"]._deps(DataDemo)
+    assert deps == {DataDemo.get_full_name(): {"n", "_scale"}}
+
+
+def test_evaluation_publishes_columns_and_returns_handle(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    handle = state.cloud
+    assert isinstance(handle, DataHandle)
+    parsed = parse_token(handle.token)
+    assert parsed is not None and parsed.kind == "data"
+    assert parsed.client_token == client_token
+    assert parsed.var_name == "cloud"
+    entry = _fresh_registry.get_columns(handle.token)
+    assert entry is not None
+    assert sorted(entry.columns) == ["mag", "x", "y"]
+    assert len(entry.columns["x"]) == 16
+
+
+def test_republish_keeps_handle_bumps_column_version(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    handle = state.cloud
+    state.n = 32
+    DataDemo.computed_vars["cloud"].mark_dirty(state)
+    assert state.cloud == handle  # stable identity, like figure vars
+    entry = _fresh_registry.get_columns(handle.token)
+    assert entry.version == 2
+    assert len(entry.columns["x"]) == 32
+
+
+def test_pre_session_short_circuit(_fresh_registry):
+    root = rx.State(_reflex_internal_init=True)
+    state = root.get_substate(tuple(DataDemo.get_full_name().split("."))[1:])
+    assert state.cloud == DataHandle("")  # data method never ran
+    assert len(_fresh_registry._column_entries) == 0
+
+
+def test_none_releases_columns(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    token = state.maybe.token
+    assert _fresh_registry.get_columns(token) is not None
+    state.n = -1
+    DataDemo.computed_vars["maybe"].mark_dirty(state)
+    assert state.maybe == DataHandle("")
+    assert _fresh_registry.get_columns(token) is None
+
+
+def test_async_variant_awaits_and_publishes(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+
+    async def main():
+        handle = await state.remote
+        assert _fresh_registry.get_columns(handle.token) is not None
+        return handle
+
+    handle = asyncio.run(main())
+    assert handle.token.startswith("xyd1|")
+
+
+def test_underscore_name_refused():
+    with pytest.raises(ValueError, match="must not start with '_'"):
+
+        class BadData(rx.State):  # noqa: F841 - definition is the assertion
+            @reflex_xy.data
+            def _hidden(self):
+                return None
+
+
+def test_method_resolvable_from_class(client_token):
+    method = builder_of(DataDemo, "cloud")
+    assert method is not None
+    state = hydrated_substate(client_token)
+    assert sorted(method(state)) == ["mag", "x", "y"]
+
+
+@pytest.mark.parametrize(
+    ("columns", "match"),
+    [
+        ([1, 2, 3], "mapping"),
+        ({1: [1.0]}, "column names must be strings"),
+        ({"x": "not-values"}, "array-like"),
+        ({"x": {"nested": 1}}, "array-like"),
+        ({"x": 3.5}, "with a length"),
+        ({"x": [1.0, 2.0], "y": [1.0]}, "share one length"),
+    ],
+)
+def test_validate_columns_rejects_malformed(columns, match):
+    with pytest.raises((TypeError, ValueError), match=match):
+        validate_columns(columns, source="Demo.cloud")
+
+
+def test_republish_rebuilds_and_bumps_mounted_dependents(_fresh_registry, client_token):
+    """The data-token -> digest index: a column republish re-binds every
+    mounted plan into a fresh figure generation (broadcast by publish)."""
+    state = hydrated_substate(client_token)
+    handle = state.cloud
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    composite = build_plan_token(plan.digest, handle.token)
+    # mount: first bind, as the namespace does on sub
+    entry = _fresh_registry.publish(
+        composite,
+        plan.bind(_fresh_registry.get_columns(handle.token).columns).figure(),
+        broadcast=False,
+    )
+    _fresh_registry.bind_plan(handle.token, plan.digest)
+    assert entry.figure.traces[0].n_points == 16
+
+    state.n = 48
+    DataDemo.computed_vars["cloud"].mark_dirty(state)
+    assert state.cloud == handle
+    rebuilt = _fresh_registry.get(composite)
+    assert rebuilt.version == entry.version + 1
+    assert rebuilt.figure.traces[0].n_points == 48
+
+
+def test_stale_column_generation_cannot_overwrite_a_newer_publish(_fresh_registry, client_token):
+    """Two concurrent republishes may finish their rebuilds in reverse order;
+    the older generation's late publish must be dropped, atomically with the
+    currency check, so subscribers never regress to stale pixels."""
+    state = hydrated_substate(client_token)
+    handle = state.cloud
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    composite = build_plan_token(plan.digest, handle.token)
+    _fresh_registry.publish(
+        composite,
+        plan.bind(_fresh_registry.get_columns(handle.token).columns).figure(),
+        broadcast=False,
+    )
+    _fresh_registry.bind_plan(handle.token, plan.digest)
+
+    old_generation = _fresh_registry.get_columns(handle.token)
+    state.n = 48
+    DataDemo.computed_vars["cloud"].mark_dirty(state)
+    assert state.cloud == handle
+    current = _fresh_registry.get(composite)
+    assert current.figure.traces[0].n_points == 48
+
+    # The older generation's rebuild finishes late: its publish must be a
+    # no-op (same for its failure path — no release, no stale err frame).
+    _fresh_registry._rebuild_dependent(handle.token, plan.digest, old_generation)
+    after = _fresh_registry.get(composite)
+    assert after is current
+    assert after.figure.traces[0].n_points == 48
+    assert after.version == current.version
+
+    broken = type(old_generation)(columns={"x": [1.0]}, token=handle.token, version=1)
+    _fresh_registry._rebuild_dependent(handle.token, plan.digest, broken)
+    assert _fresh_registry.get(composite) is current  # stale failure: no release
+
+
+def test_republish_drops_unmounted_dependents_from_the_index(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    handle = state.cloud
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    _fresh_registry.bind_plan(handle.token, plan.digest)  # indexed, never mounted
+
+    state.n = 20
+    DataDemo.computed_vars["cloud"].mark_dirty(state)
+    assert state.cloud == handle
+    assert _fresh_registry._digests_by_data_token.get(handle.token) is None
+
+
+def test_release_columns_releases_dependent_figures(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    handle = state.maybe.token
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    composite = build_plan_token(plan.digest, handle)
+    columns = _fresh_registry.get_columns(handle).columns
+    _fresh_registry.publish(composite, plan.bind(columns).figure(), broadcast=False)
+    _fresh_registry.bind_plan(handle, plan.digest)
+
+    state.n = -1
+    DataDemo.computed_vars["maybe"].mark_dirty(state)
+    assert state.maybe == DataHandle("")
+    assert _fresh_registry.get(composite) is None
+    assert _fresh_registry.get_columns(handle) is None
+
+
+def test_column_entries_sweep_like_figures(_fresh_registry, client_token):
+    state = hydrated_substate(client_token)
+    token = state.cloud.token
+    entry = _fresh_registry.get_columns(token)
+    dropped = _fresh_registry.sweep(now=entry.last_access + 10**9)
+    assert token in dropped
+    assert _fresh_registry.get_columns(token) is None
diff --git a/tests/reflex_adapter/test_plan.py b/tests/reflex_adapter/test_plan.py
new file mode 100644
index 00000000..22bddae2
--- /dev/null
+++ b/tests/reflex_adapter/test_plan.py
@@ -0,0 +1,132 @@
+"""Chart plans: compile-time validation, content addressing, binding."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+import xy
+from reflex_xy.plan import (
+    PLAN_VERSION,
+    PlanBindError,
+    PlanError,
+    PlanMissError,
+    build_plan,
+    plan_of,
+    require_plan,
+)
+
+
+def scatter_plan(**mark_options):
+    return build_plan(
+        "scatter_chart",
+        (xy.scatter("x", "y", color="mag", **mark_options), xy.x_axis(label="sigma")),
+        {"title": "cloud"},
+    )
+
+
+def test_plan_records_probed_columns_in_resolution_order():
+    plan = scatter_plan()
+    assert plan.columns == ("x", "y", "mag")
+
+
+def test_digest_is_stable_across_identical_builds():
+    """Content addressing: separately constructed identical trees agree, so
+    every worker evaluating the same page derives the same digest (X4)."""
+    assert scatter_plan().digest == scatter_plan().digest
+    assert scatter_plan().digest != scatter_plan(opacity=0.5).digest
+
+
+def test_digest_golden_pins_the_plan_format():
+    """plan_version is part of the serialization: accidental format churn
+    shows up here as a digest change. An *intentional* format change bumps
+    PLAN_VERSION and re-records this golden (old digests are content
+    addresses — stale subscribers resync, nothing migrates)."""
+    assert PLAN_VERSION == 1
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    assert plan.digest == "b7d0b4245b686130e37d"
+
+
+def test_zero_row_probe_fires_the_full_validation_gate():
+    with pytest.raises(ValueError, match="colormap"):
+        scatter_plan(colormap="virids")
+    with pytest.raises(ValueError, match="symbol"):
+        scatter_plan(symbol="marsian")
+    # unresolved axis-id references are figure-compile errors too (X2)
+    with pytest.raises(ValueError, match="axis"):
+        build_plan("scatter_chart", (xy.scatter("x", "y", y_axis="y2"),), {})
+
+
+def test_plans_refuse_concrete_arrays():
+    with pytest.raises(PlanError, match="data-free"):
+        build_plan("scatter_chart", (xy.scatter(np.array([1.0]), np.array([2.0])),), {})
+
+
+def test_plans_refuse_per_mark_data():
+    with pytest.raises(PlanError, match="per-mark data="):
+        build_plan("scatter_chart", (xy.scatter("x", "y", data={"x": [], "y": []}),), {})
+
+
+def test_plans_refuse_render_components():
+    with pytest.raises(PlanError, match="render"):
+        build_plan(
+            "scatter_chart",
+            (xy.scatter("x", "y"), xy.legend(render=object())),
+            {},
+        )
+
+
+def test_plan_is_a_snapshot_immune_to_later_node_mutation():
+    """The registered plan holds a deep copy of what was hashed: mutating a
+    reused mark node or the chart props afterwards must not change binding
+    behavior behind an unchanged digest/columns record."""
+    mark = xy.scatter("x", "y", color="mag")
+    props = {"title": "cloud"}
+    plan = build_plan("scatter_chart", (mark,), props)
+    assert plan.columns == ("x", "y", "mag")
+
+    mark.props["color"] = "sneaky"  # node reused and mutated by page code
+    props["title"] = "renamed"
+    assert plan.children[0] is not mark
+    assert plan.children[0].props["color"] == "mag"
+    assert plan.chart_props == {"title": "cloud"}
+    # binding still resolves the snapshot's channels, not the mutated node's
+    fig = plan.bind({"x": [1.0], "y": [2.0], "mag": [3.0]}).figure()
+    assert fig.traces[0].n_points == 1
+
+
+def test_bind_produces_fresh_figures_per_call():
+    """X3: Chart.figure() memoizes, so bind() must mint a fresh Chart —
+    two binds with different columns give independent figures."""
+    plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+    small = plan.bind({"x": [1.0], "y": [2.0]}).figure()
+    large = plan.bind({"x": [1.0, 2.0, 3.0], "y": [2.0, 3.0, 4.0]}).figure()
+    assert small is not large
+    assert small.traces[0].n_points == 1
+    assert large.traces[0].n_points == 3
+
+
+def test_bind_error_names_both_sides():
+    plan = scatter_plan()
+    with pytest.raises(
+        PlanBindError, match=r"plan binds column 'mag'; Dash.cloud produced \{x, y\}"
+    ):
+        plan.bind({"x": [1.0], "y": [2.0]}, source="Dash.cloud")
+
+
+def test_probe_collects_the_tailwind_inventory():
+    plan = build_plan(
+        "scatter_chart",
+        (xy.scatter("x", "y"), xy.legend(class_name="max-h-24 overflow-y-auto")),
+        {"class_name": "rounded-xl"},
+    )
+    assert "rounded-xl" in plan.tailwind_classes
+    assert "max-h-24" in plan.tailwind_classes
+
+
+def test_registry_lookup_and_miss():
+    plan = scatter_plan()
+    assert plan_of(plan.digest) is plan
+    assert plan_of("feedfacefeedfacefeed") is None
+    with pytest.raises(PlanMissError, match="feedfacefeedfacefeed"):
+        require_plan("feedfacefeedfacefeed")