From d26e584cc58ffaf7fdce534444bd2493117f4ea8 Mon Sep 17 00:00:00 2001
From: Farhan Ali Raza <62690310+FarhanAliRaza@users.noreply.github.com>
Date: Wed, 5 Aug 2026 19:15:29 +0500
Subject: [PATCH 1/3] feat(reflex): serve composite plan tokens over the data
 plane
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Makes xyp1|<digest>|<data token> a servable figure identity. The namespace
learns one concept — what a token reveals about affinity and
rebuildability — instead of branching on prefixes at each call site: a
composite enforces affinity through its embedded data token and rebuilds
like any state token; a bare xyd1 token names columns, never a figure, so
it keeps affinity but is never served or rebuilt as one.

Serving a composite is plan lookup + columns (registry hit, else the data
method re-run against session state) + bind into a fresh Chart. Both halves
are independently recoverable on any worker, so §3.2's reconnect promise
holds for this tier without a central store.

Failures are typed rather than flattened to "unknown figure token": a plan
miss (hot-reload digest drift) answers err {resync} naming the digest, and
a bind mismatch answers the reason, naming both sides. The registry's
error seam is wired to a room-wide err broadcast here — a column republish
whose bind fails has no request to answer, and without it subscribers would
sit on stale pixels with nothing in the log.

The wire envelope grew no fields: rooms, versions, mid addressing, and the
attachment cap treat a composite as an ordinary fig string.

Spec: reflex-integration.md §3.6 (composite tokens, republish fan-out),
file map.
---
 python/reflex_xy/app.py                       |   1 +
 python/reflex_xy/namespace.py                 | 113 ++++++--
 python/reflex_xy/state_bridge.py              | 103 +++++--
 spec/design/reflex-integration.md             |  35 ++-
 tests/reflex_adapter/conftest.py              |   5 +-
 .../reflex_adapter/test_socket_data_plane.py  | 253 +++++++++++++++++-
 6 files changed, 452 insertions(+), 58 deletions(-)

diff --git a/python/reflex_xy/app.py b/python/reflex_xy/app.py
index 12542c2e..8d285a64 100644
--- a/python/reflex_xy/app.py
+++ b/python/reflex_xy/app.py
@@ -67,6 +67,7 @@ def wire(namespace: XYNamespace) -> None:
     """Point the registry's fan-out seams at a namespace (setup and tests)."""
     registry.on_publish(namespace.broadcast_payload)
     registry.on_push(namespace.broadcast_message)
+    registry.on_error(namespace.broadcast_error)
 
 
 async def _xy_lifespan() -> None:
diff --git a/python/reflex_xy/namespace.py b/python/reflex_xy/namespace.py
index 993448ac..8a273243 100644
--- a/python/reflex_xy/namespace.py
+++ b/python/reflex_xy/namespace.py
@@ -39,14 +39,16 @@
 import asyncio
 import urllib.parse
 from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Optional
 
 from socketio import AsyncNamespace
 
 from xy.channel import handle_message
 
+from .plan import PlanError, PlanMissError
 from .registry import FigureEntry, FigureRegistry
-from .tokens import parse_token
+from .tokens import parse_plan_token, parse_token
 
 if TYPE_CHECKING:
     from xy._figure import Figure
@@ -104,9 +106,45 @@ def _handle_entry_message(entry: FigureEntry, content: Any) -> Any:
 
 # An async callable(token) -> Figure | None: given a parseable figure token,
 # rebuild the figure from Reflex state (wired by app.setup; see state_bridge).
+# May raise PlanError subclasses for spec-aware err frames.
 RebuildHook = Callable[[str], Awaitable[Optional["Figure"]]]
 
 
+@dataclass(frozen=True)
+class _RebuildFailure:
+    """Client-facing outcome of a failed rebuild attempt."""
+
+    error: str
+    resync: bool = False
+
+
+@dataclass(frozen=True)
+class _TokenIdentity:
+    """What a figure token reveals: session affinity and rebuildability."""
+
+    affinity_client: Optional[str]
+    rebuildable: bool
+    plan_bound: Optional[tuple[str, str]] = None  # (data_token, digest)
+
+
+def _token_identity(token: str) -> _TokenIdentity:
+    composite = parse_plan_token(token)
+    if composite is not None:
+        return _TokenIdentity(
+            affinity_client=composite.data.client_token,
+            rebuildable=True,
+            plan_bound=(composite.data_token, composite.digest),
+        )
+    parsed = parse_token(token)
+    if parsed is not None and parsed.kind == "figure":
+        return _TokenIdentity(affinity_client=parsed.client_token, rebuildable=True)
+    if parsed is not None:
+        # A bare data token names columns, never a figure: enforce affinity
+        # (it embeds a session) but never serve or rebuild it as one.
+        return _TokenIdentity(affinity_client=parsed.client_token, rebuildable=False)
+    return _TokenIdentity(affinity_client=None, rebuildable=False)
+
+
 def _plain(value: Any) -> Any:
     """Best-effort JSON-safe copy for small reply metadata.
 
@@ -159,7 +197,7 @@ def __init__(
         # room fan-out, so one cancelled waiter cannot cancel the attempt and
         # a failed builder is not rerun serially by every existing waiter.
         self._rebuild_attempts: dict[
-            str, asyncio.Task[tuple[Optional[FigureEntry], Optional[str]]]
+            str, asyncio.Task[tuple[Optional[FigureEntry], Optional[_RebuildFailure]]]
         ] = {}
 
     # -- connection lifecycle ------------------------------------------------
@@ -208,7 +246,13 @@ async def on_sub(self, sid: str, data: Any) -> None:
                 # path has proved the SID is still live. A concurrent
                 # disconnect can now only remove this record, never precede
                 # and be undone by it.
-                self.registry.subscribe(token, sid, rebuildable=parse_token(token) is not None)
+                identity = _token_identity(token)
+                self.registry.subscribe(token, sid, rebuildable=identity.rebuildable)
+                if identity.plan_bound is not None:
+                    # Index the mounted plan before serving, so a column
+                    # republish racing this subscribe rebuilds it (the
+                    # re-read below then serves that fresher generation).
+                    self.registry.bind_plan(*identity.plan_bound)
                 # A normal state publish can replace a just-rebuilt entry
                 # while its room-wide broadcast is still completing, before
                 # this SID joins. Re-read after the join: replacements before
@@ -438,7 +482,7 @@ def _release_subscription_lock(self, key: tuple[str, str], lock: asyncio.Lock) -
 
     def _start_rebuild_attempt(
         self, token: str
-    ) -> asyncio.Task[tuple[Optional[FigureEntry], Optional[str]]]:
+    ) -> asyncio.Task[tuple[Optional[FigureEntry], Optional[_RebuildFailure]]]:
         """Start and retain one shared cache-miss attempt for ``token``."""
         # Install the guard before publishing the task in ``_rebuild_attempts``.
         # A concurrent waiter can then distinguish this live attempt from one
@@ -447,7 +491,9 @@ def _start_rebuild_attempt(
         task = asyncio.create_task(self._run_rebuild_attempt(token, entry, guard))
         self._rebuild_attempts[token] = task
 
-        def forget(completed: asyncio.Task[tuple[Optional[FigureEntry], Optional[str]]]) -> None:
+        def forget(
+            completed: asyncio.Task[tuple[Optional[FigureEntry], Optional[_RebuildFailure]]],
+        ) -> None:
             if self._rebuild_attempts.get(token) is completed:
                 self._rebuild_attempts.pop(token, None)
 
@@ -459,19 +505,26 @@ async def _run_rebuild_attempt(
         token: str,
         entry: Optional[FigureEntry],
         guard: Optional[object],
-    ) -> tuple[Optional[FigureEntry], Optional[str]]:
+    ) -> tuple[Optional[FigureEntry], Optional[_RebuildFailure]]:
         """Build, conditionally publish, and fan out one total rebuild attempt."""
+        unknown = _RebuildFailure("unknown figure token")
         if entry is not None:
             return entry, None
         if guard is None:  # defensive: a miss always receives one bounded guard
-            return None, "unknown figure token"
+            return None, unknown
 
         try:
             rebuild = self._rebuild
             if rebuild is None:
-                return None, "unknown figure token"
+                return None, unknown
             try:
                 figure = await rebuild(token)
+            except PlanError as exc:
+                # Spec-aware failures of the data-bound tier: a stale plan
+                # digest asks the client to resync (the recompiled page
+                # carries the new digest); a bind mismatch names both sides
+                # and is not retryable as-is.
+                return None, _RebuildFailure(str(exc), resync=isinstance(exc, PlanMissError))
             except Exception:  # noqa: BLE001 - user builder code is an input boundary
                 figure = None
 
@@ -480,12 +533,12 @@ async def _run_rebuild_attempt(
                 # awaiting. Use it instead of reporting a stale rebuild failure.
                 entry = self.registry.get(token)
                 if entry is None:
-                    return None, "unknown figure token"
+                    return None, unknown
                 return entry, None
 
             entry, inserted = self.registry.publish_if_missing(token, figure, guard=guard)
             if entry is None:
-                return None, "unknown figure token"
+                return None, unknown
             if not inserted:
                 return entry, None
 
@@ -499,7 +552,7 @@ async def _run_rebuild_attempt(
                     current = self.registry.get(token)
                     if current is not None:
                         return current, None
-                return None, "rebuild failed"
+                return None, _RebuildFailure("rebuild failed")
 
             # ``broadcast_payload`` intentionally no-ops when its generation
             # went stale. Resolve that race explicitly: a replacement wins,
@@ -507,7 +560,7 @@ async def _run_rebuild_attempt(
             if not self.registry.is_current(token, entry):
                 current = self.registry.get(token)
                 if current is None:
-                    return None, "unknown figure token"
+                    return None, unknown
                 return current, None
             return entry, None
         finally:
@@ -527,10 +580,10 @@ async def _entry_for(
         token = self._token_of(data)
         if token is None:
             return None, None, False
-        parsed = parse_token(token)
-        if parsed is not None:
+        identity = _token_identity(token)
+        if identity.affinity_client is not None:
             session = await self.get_session(sid)
-            if session.get("client_token") != parsed.client_token:
+            if session.get("client_token") != identity.affinity_client:
                 await self._err(sid, token, "figure belongs to another session")
                 return token, None, False
         entry, rebuild_guarded = self.registry.get_with_rebuild_guard(token)
@@ -541,7 +594,12 @@ async def _entry_for(
         # forever for older user rebuild code. An entry whose guard is still
         # valid remains provisional until its rebuild fan-out completes.
         initially_missing = entry is None or (attempt is not None and rebuild_guarded)
-        if parsed is not None and allow_rebuild and self._rebuild is not None and initially_missing:
+        if (
+            identity.rebuildable
+            and allow_rebuild
+            and self._rebuild is not None
+            and initially_missing
+        ):
             # The task spans builder, conditional insertion, and existing-room
             # fan-out. All requests that observed this in-flight miss share its
             # result and must drop pre-payload interactions, even when another
@@ -553,14 +611,27 @@ async def _entry_for(
                 attempt = None
             if attempt is None:
                 attempt = self._start_rebuild_attempt(token)
-            entry, error = await asyncio.shield(attempt)
-            if error is not None:
-                await self._err(sid, token, error)
+            entry, failure = await asyncio.shield(attempt)
+            if failure is not None:
+                await self._err(sid, token, failure.error, resync=failure.resync)
                 return token, None, initially_missing
         if entry is None:
             await self._err(sid, token, "unknown figure token")
             return token, None, False
         return token, entry, initially_missing
 
-    async def _err(self, sid: str, token: Optional[str], error: str) -> None:
-        await self.emit("err", {"fig": token, "error": error}, to=sid)
+    async def _err(
+        self, sid: str, token: Optional[str], error: str, *, resync: bool = False
+    ) -> None:
+        envelope: dict[str, Any] = {"fig": token, "error": error}
+        if resync:
+            envelope["resync"] = True
+        await self.emit("err", envelope, to=sid)
+
+    async def broadcast_error(self, token: str, error: str, resync: bool = False) -> None:
+        """Room-wide err frame for server-side failures with no request to
+        answer (e.g. a column republish whose plan bind fails)."""
+        envelope: dict[str, Any] = {"fig": token, "error": error}
+        if resync:
+            envelope["resync"] = True
+        await self.emit("err", envelope, room=self._room(token))
diff --git a/python/reflex_xy/state_bridge.py b/python/reflex_xy/state_bridge.py
index 2eb0d0a4..e76d03af 100644
--- a/python/reflex_xy/state_bridge.py
+++ b/python/reflex_xy/state_bridge.py
@@ -1,17 +1,25 @@
-"""Rebuild figures from Reflex state: the distributed-deployment answer.
+"""Rebuild figures and datasets from Reflex state: the distributed answer.
 
 The figure registry is process-local. What makes that safe in a
-multi-worker / reconnecting world is this module: given a state token
-(`xyv1|client|state|var`) and the app's state manager, we can always
-recover the figure by re-running the builder against the session's state —
-which Reflex already stores durably (memory/disk/redis) and already knows
-how to hand to any worker. No figure server, no data in Redis beyond the
-state that was there anyway (§27 applied to processes: the figure is a
-rebuildable cache, Reflex state is canonical).
+multi-worker / reconnecting world is this module: given a state token, we
+can always recover the served object by re-running the decorated state
+method against the session's state — which Reflex already stores durably
+(memory/disk/redis) and already knows how to hand to any worker. No figure
+server, no data in Redis beyond the state that was there anyway (§27
+applied to processes: the served object is a rebuildable cache, Reflex
+state is canonical).
+
+Three recipes, one contract:
+
+- `xyv1|client|state|var` — re-run the `@reflex_xy.figure` builder → Figure.
+- `xyd1|client|state|var` — re-run the `@reflex_xy.data` method → columns.
+- `xyp1|digest|xyd1|…` — plan lookup (process-local; every worker's map
+  is populated at startup, see app.py `_ensure_page_plans`) + column
+  resolve (registry hit, else the xyd1 recipe) + bind → Figure.
 
 Read-only by design: rebuilds use `state_manager.get_state` (no state lock,
-no delta emission). Builders must therefore be pure functions of state —
-the same contract cached computed vars already impose.
+no delta emission). Builders and data methods must therefore be pure
+functions of state — the same contract cached computed vars already impose.
 """
 
 from __future__ import annotations
@@ -19,13 +27,15 @@
 import inspect
 from typing import TYPE_CHECKING, Any, Optional
 
-from .registry import _figure_of
-from .tokens import ParsedToken, builder_of, parse_token
+from .data_vars import validate_columns
+from .plan import require_plan
+from .registry import _figure_of, registry
+from .tokens import ParsedPlanToken, ParsedToken, builder_of, parse_plan_token, parse_token
 
 if TYPE_CHECKING:
     from xy._figure import Figure
 
-__all__ = ["make_rebuild_hook", "rebuild_figure"]
+__all__ = ["make_rebuild_hook", "rebuild_data", "rebuild_figure", "rebuild_plan_figure"]
 
 
 def _resolve_state_cls(state_full_name: str) -> Any:
@@ -40,40 +50,81 @@ def _resolve_state_cls(state_full_name: str) -> Any:
     return rx.State.get_class_substate(tuple(state_full_name.split(".")))
 
 
-async def rebuild_figure(app: Any, parsed: ParsedToken) -> Optional["Figure"]:
-    """Re-run a figure var's builder against the session's stored state."""
+async def _run_state_method(app: Any, parsed: ParsedToken) -> Any:
+    """Re-run a decorated state method against the session's stored state."""
     import reflex as rx
 
-    from xy._figure import Figure
-
     try:
         state_cls = _resolve_state_cls(parsed.state_full_name)
     except (KeyError, ValueError):
         return None
-    builder = builder_of(state_cls, parsed.var_name)
-    if builder is None:
+    method = builder_of(state_cls, parsed.var_name)
+    if method is None:
         return None
     token = rx.BaseStateToken(ident=parsed.client_token, cls=rx.State)
     root = await app.state_manager.get_state(token)
     substate = await root.get_state(state_cls)
-    # Async builders (AsyncFigureVar) await their data source here exactly
-    # as they would during normal var evaluation.
-    if inspect.iscoroutinefunction(builder):
-        chart = await builder(substate)
-    else:
-        chart = builder(substate)
+    # Async builders/data methods await their data source here exactly as
+    # they would during normal var evaluation.
+    if inspect.iscoroutinefunction(method):
+        return await method(substate)
+    return method(substate)
+
+
+async def rebuild_figure(app: Any, parsed: ParsedToken) -> Optional["Figure"]:
+    """Re-run a figure var's builder against the session's stored state."""
+    from xy._figure import Figure
+
+    if parsed.kind != "figure":
+        return None
+    chart = await _run_state_method(app, parsed)
     if chart is None:
         return None
     figure = _figure_of(chart)
     return figure if isinstance(figure, Figure) else None
 
 
+async def rebuild_data(app: Any, parsed: ParsedToken) -> Optional[dict[str, Any]]:
+    """Re-run a data var's method against the session's stored state."""
+    if parsed.kind != "data":
+        return None
+    columns = await _run_state_method(app, parsed)
+    if columns is None:
+        return None
+    return validate_columns(
+        columns, source=f"{parsed.state_full_name.rsplit('.', 1)[-1]}.{parsed.var_name}"
+    )
+
+
+async def rebuild_plan_figure(app: Any, composite: ParsedPlanToken) -> Optional["Figure"]:
+    """Recover a data-bound figure: plan (local map) + columns (registry or
+    state) + bind. Raises PlanMissError / PlanBindError for spec-aware `err`
+    frames; anything else fails closed in the namespace."""
+    plan = require_plan(composite.digest)
+    entry = registry.get_columns(composite.data_token)
+    if entry is not None:
+        columns = entry.columns
+    else:
+        columns = await rebuild_data(app, composite.data)
+        if columns is None:
+            return None
+        # Future binds (other plans over the same data var) hit the cache;
+        # bind_plan below is the subscribe path's job, not the rebuild's.
+        registry.publish_columns(composite.data_token, columns)
+    source = f"{composite.data.state_full_name.rsplit('.', 1)[-1]}.{composite.data.var_name}"
+    return plan.bind(columns, source=source).figure()
+
+
 def make_rebuild_hook(app: Any) -> Any:
     """The namespace's RebuildHook, bound to one app instance."""
 
     async def _rebuild(token_str: str) -> Optional["Figure"]:
+        composite = parse_plan_token(token_str)
+        if composite is not None:
+            return await rebuild_plan_figure(app, composite)
         parsed = parse_token(token_str)
-        if parsed is None:
+        if parsed is None or parsed.kind != "figure":
+            # Bare data tokens never serve figures; fail closed.
             return None
         return await rebuild_figure(app, parsed)
 
diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md
index 4f97b6cf..a3ed2187 100644
--- a/spec/design/reflex-integration.md
+++ b/spec/design/reflex-integration.md
@@ -491,7 +491,31 @@ 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.
+request.
+
+**Composite tokens.** The two halves compose into one figure identity —
+`xyp1|<digest>|<xyd1 token>` — subscribed as a unit. Rooms, versions, `mid`
+addressing, the attachment cap, and every message below the subscribe path
+treat it as an ordinary `fig` string; the envelope grew no fields. Serving
+it = `plan_of(digest)` + columns (registry hit, else `rebuild_data` re-runs
+the data method against session state) + bind into a fresh `Chart` →
+`figure()` → cached under the composite token as a normal, TTL-sweepable
+`FigureEntry`. The rebuild recipe is (plan map, data method): both halves
+independently recoverable on any worker, which is what keeps §3.2's
+promise intact for this tier. Session affinity reads the embedded data
+token, so a composite belonging to another session is refused exactly like
+a figure token; a *bare* `xyd1` token names columns, never a figure, and is
+refused as a subscription outright.
+
+**Republish fan-out.** A data var recompute publishes new columns and
+re-binds every mounted dependent through the `data token → {digests}`
+index — fresh figures, bumped versions, coalesced room broadcasts, exactly
+the figure-var republish machinery. The index is added to when a composite
+binds and pruned when a republish finds the plan unmounted, so it stays
+bounded by mounted plans. Failure stays loud: a bind that stops matching
+(possible only for untyped data vars) logs server-side, releases the
+composite entry, and answers the room `err {resync}`; a stale digest
+(hot-reload drift) answers `err {resync}` naming the digest.
 
 **Republish ordering.** Dependent rebuilds run outside the registry mutex
 (they execute user-scale figure builds), so two republishes of one data
@@ -821,9 +845,11 @@ python/reflex_xy/
   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
+  state_bridge.py            token -> state_manager -> builder/data/plan
+                             rebuild hooks
   namespace.py               XYNamespace: sub/unsub/msg, payload/msg/err,
-                             affinity, rebuild-on-miss, binary attachments
+                             affinity (incl. composite), rebuild-on-miss,
+                             binary attachments
   app.py                     setup(app), XYPlugin (post_compile), lifespan
   component.py               chart(figure=...) -> rx.Component (local-JSX
                              library); typed figure prop; static tier
@@ -849,7 +875,8 @@ tests/reflex_adapter/        token/registry/var/data-var/plan/bridge/
                              contract pins (R1/R7/R8), and a real-websocket
                              integration suite (uvicorn + socketio client)
                              covering payload/pick/select/affinity/rebuild/
-                             publish-broadcast/append/unsub
+                             publish-broadcast/append/unsub + the composite
+                             plan tier (fan-out, plan-miss resync, bind errs)
 ```
 
 `inline()` (content-addressed pinned tokens, §3.4) lives in the package
diff --git a/tests/reflex_adapter/conftest.py b/tests/reflex_adapter/conftest.py
index f52bbee3..344a5410 100644
--- a/tests/reflex_adapter/conftest.py
+++ b/tests/reflex_adapter/conftest.py
@@ -14,16 +14,19 @@
 pytest.importorskip("reflex_xy")
 
 import reflex_xy.app as adapter_app  # noqa: E402
+from reflex_xy.plan import reset_plans_for_tests  # noqa: E402
 from reflex_xy.registry import reset_registry_for_tests  # noqa: E402
 
 
 @pytest.fixture(autouse=True)
 def _fresh_registry():
-    """Isolate registry + wiring between tests."""
+    """Isolate registry + plan map + wiring between tests."""
     registry = reset_registry_for_tests()
+    reset_plans_for_tests()
     adapter_app.reset_setup_for_tests()
     yield registry
     reset_registry_for_tests()
+    reset_plans_for_tests()
     adapter_app.reset_setup_for_tests()
 
 
diff --git a/tests/reflex_adapter/test_socket_data_plane.py b/tests/reflex_adapter/test_socket_data_plane.py
index 39f74419..3db9c720 100644
--- a/tests/reflex_adapter/test_socket_data_plane.py
+++ b/tests/reflex_adapter/test_socket_data_plane.py
@@ -17,18 +17,24 @@
 import socket
 import threading
 from types import SimpleNamespace
+from typing import TypedDict
 
 import numpy as np
 import pytest
+import reflex as rx
 import socketio
 import uvicorn
+from reflex.istate.manager.memory import StateManagerMemory
 from reflex_base.utils import format as reflex_format
 
+import reflex_xy
 import xy
 from reflex_xy.app import wire
 from reflex_xy.namespace import XYNamespace
+from reflex_xy.plan import build_plan
 from reflex_xy.registry import registry
-from reflex_xy.tokens import build_state_token
+from reflex_xy.state_bridge import make_rebuild_hook
+from reflex_xy.tokens import build_data_token, build_plan_token, build_state_token
 
 CLIENT_TOKEN = "11111111-2222-4333-8444-555566667777"
 OTHER_TOKEN = "99999999-8888-4777-8666-555544443333"
@@ -758,7 +764,7 @@ async def rebuild(token_str):
     async def get_session(sid):
         return {"client_token": CLIENT_TOKEN}
 
-    async def send_error(sid, token, error):
+    async def send_error(sid, token, error, resync=False):
         errors.append((sid, token, error))
 
     async def broadcast(token, entry):
@@ -835,7 +841,7 @@ async def broadcast(token, entry):
         if len(broadcasts) == 1:
             raise RuntimeError("transport failed")
 
-    async def send_error(sid, token, error):
+    async def send_error(sid, token, error, resync=False):
         errors.append((sid, token, error))
 
     async def enter_room(sid, room):
@@ -908,7 +914,7 @@ async def fail_after_replacement(token, entry):
         replacement = registry.publish(token, replacement_figure, broadcast=False)
         raise RuntimeError("old generation transport failed")
 
-    async def send_error(sid, token, error):
+    async def send_error(sid, token, error, resync=False):
         errors.append((sid, token, error))
 
     monkeypatch.setattr(namespace, "get_session", get_session)
@@ -971,7 +977,7 @@ async def fail_then_deliver_from_publish(token, entry):
         assert registry._rebuildable_subscribers[state_token] == {"sid-existing"}
         delivered.set()
 
-    async def send_error(sid, token, error):
+    async def send_error(sid, token, error, resync=False):
         errors.append((sid, token, error))
 
     monkeypatch.setattr(namespace, "get_session", get_session)
@@ -1176,7 +1182,7 @@ async def rebuild(token_str):
     async def get_session(sid):
         return {"client_token": CLIENT_TOKEN}
 
-    async def send_error(sid, token, error):
+    async def send_error(sid, token, error, resync=False):
         errors.append((sid, token, error))
 
     async def broadcast(token, entry):
@@ -1648,3 +1654,238 @@ async def main():
         assert replacement.version == 1
 
     run(main())
+
+
+# -- the data-bound (plan) tier over the same wire ---------------------------
+#
+# Composite figure identity xyp1|<digest>|<xyd1 token>: rooms, versioning,
+# mid addressing, and the attachment-cap logic are reused unchanged — the
+# composite token is just another `fig` string to everything below the
+# subscribe path.
+
+
+class PlaneSchema(TypedDict):
+    x: np.ndarray
+    y: np.ndarray
+
+
+class PlaneData(rx.State):
+    n: int = 24
+
+    @reflex_xy.data
+    def table(self) -> PlaneSchema:
+        xs = np.linspace(0.0, 1.0, self.n)
+        return {"x": xs, "y": xs * 2.0}
+
+
+def make_plane_app():
+    return SimpleNamespace(state_manager=StateManagerMemory())
+
+
+def plane_plan():
+    return build_plan("scatter_chart", (xy.scatter("x", "y"),), {})
+
+
+def plane_data_token(client_token: str = CLIENT_TOKEN) -> str:
+    return build_data_token(client_token, PlaneData.get_full_name(), "table")
+
+
+def test_composite_sub_serves_bound_payload_and_interactions(_fresh_registry):
+    """sub/msg on a composite token: plan lookup + column resolve + bind,
+    then the ordinary payload/pick machinery."""
+    app = make_plane_app()
+
+    async def main():
+        plan = plane_plan()
+        data_token = plane_data_token()
+        composite = build_plan_token(plan.digest, data_token)
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            payload = await collector.next(collector.payloads)
+            assert payload["fig"] == composite
+            assert payload["spec"]["traces"][0]["n_points"] == 24
+            # interactions round-trip against the bound figure
+            await client.emit(
+                "msg",
+                {
+                    "fig": composite,
+                    "v": payload["version"],
+                    "mid": "m1",
+                    "m": {"type": "pick", "trace": 0, "index": 3, "seq": "pick:1"},
+                },
+                namespace="/_xy",
+            )
+            reply = await collector.next(collector.messages)
+            assert reply["message"]["type"] == "pick_result"
+            await client.disconnect()
+        # both halves are cached now: columns and the bound figure
+        assert registry.get_columns(data_token) is not None
+        assert registry.get(composite) is not None
+
+    run(main())
+
+
+def test_composite_rebuild_reads_session_state(_fresh_registry):
+    """The data half is rebuilt through the state bridge (mutated session
+    state, not defaults) when neither half is cached."""
+    app = make_plane_app()
+    token_obj = rx.BaseStateToken(ident=CLIENT_TOKEN, cls=rx.State)
+
+    async def main():
+        async with app.state_manager.modify_state(token_obj) as root:
+            sub = await root.get_state(PlaneData)
+            sub.n = 7
+        plan = plane_plan()
+        composite = build_plan_token(plan.digest, plane_data_token())
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            payload = await collector.next(collector.payloads)
+            assert payload["spec"]["traces"][0]["n_points"] == 7
+            await client.disconnect()
+
+    run(main())
+
+
+def test_column_republish_fans_out_to_every_dependent_plan(_fresh_registry):
+    """One data var, two mounted plans: a republish rebuilds and broadcasts
+    both composite figures."""
+    app = make_plane_app()
+
+    async def main():
+        scatter_plan = plane_plan()
+        line_plan = build_plan("line_chart", (xy.line("x", "y"),), {})
+        data_token = plane_data_token()
+        scatter_fig = build_plan_token(scatter_plan.digest, data_token)
+        line_fig = build_plan_token(line_plan.digest, data_token)
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": scatter_fig, "mid": "m1"}, namespace="/_xy")
+            first = await collector.next(collector.payloads)
+            await client.emit("sub", {"fig": line_fig, "mid": "m2"}, namespace="/_xy")
+            second = await collector.next(collector.payloads)
+            assert {first["fig"], second["fig"]} == {scatter_fig, line_fig}
+
+            # the data var recomputes (as a state delta evaluation would)
+            registry.publish_columns(
+                data_token,
+                {"x": np.linspace(0.0, 1.0, 5), "y": np.linspace(0.0, 1.0, 5)},
+            )
+            refreshed = {}
+            for _ in range(2):
+                payload = await collector.next(collector.payloads)
+                refreshed[payload["fig"]] = payload["spec"]["traces"][0]["n_points"]
+            assert refreshed == {scatter_fig: 5, line_fig: 5}
+            await client.disconnect()
+
+    run(main())
+
+
+def test_composite_affinity_uses_the_embedded_data_client(_fresh_registry):
+    app = make_plane_app()
+
+    async def main():
+        plan = plane_plan()
+        composite = build_plan_token(plan.digest, plane_data_token(CLIENT_TOKEN))
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            thief = await connect_client(url, client_token=OTHER_TOKEN)
+            thief_collector = Collector(thief)
+            await thief.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            error = await thief_collector.next(thief_collector.errors)
+            assert "another session" in error["error"]
+            await thief.disconnect()
+
+    run(main())
+
+
+def test_plan_miss_answers_err_resync_naming_the_digest(_fresh_registry):
+    """Hot-reload drift: a subscriber holding a stale digest is asked to
+    resync (the recompiled page carries the new digest)."""
+    app = make_plane_app()
+
+    async def main():
+        composite = build_plan_token("feedfacefeedfacefeed", plane_data_token())
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            error = await collector.next(collector.errors)
+            assert "feedfacefeedfacefeed" in error["error"]
+            assert error["resync"] is True
+            await client.disconnect()
+
+    run(main())
+
+
+def test_bind_mismatch_answers_err_without_resync(_fresh_registry):
+    """An untyped data var producing the wrong columns: the err frame names
+    both sides and does not ask for a pointless resync."""
+
+    class MismatchData(rx.State):
+        @reflex_xy.data
+        def rows(self):
+            return {"only": [1.0, 2.0]}
+
+    app = make_plane_app()
+
+    async def main():
+        plan = plane_plan()
+        data_token = build_data_token(CLIENT_TOKEN, MismatchData.get_full_name(), "rows")
+        composite = build_plan_token(plan.digest, data_token)
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            error = await collector.next(collector.errors)
+            assert "plan binds" in error["error"]
+            assert "'x'" in error["error"]
+            assert error.get("resync") is None
+            await client.disconnect()
+
+    run(main())
+
+
+def test_republish_bind_failure_broadcasts_err_and_releases(_fresh_registry):
+    """A republish whose columns stop satisfying a mounted plan must not
+    freeze subscribers silently: the composite entry is released and the
+    room gets an err frame asking for a bounded resync."""
+    app = make_plane_app()
+
+    async def main():
+        plan = plane_plan()
+        data_token = plane_data_token()
+        composite = build_plan_token(plan.digest, data_token)
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": composite, "mid": "m1"}, namespace="/_xy")
+            await collector.next(collector.payloads)
+
+            registry.publish_columns(data_token, {"wrong": [1.0]})
+            error = await collector.next(collector.errors)
+            assert "plan binds" in error["error"]
+            assert error["resync"] is True
+            assert registry.get(composite) is None
+            await client.disconnect()
+
+    run(main())
+
+
+def test_bare_data_token_is_not_a_figure(_fresh_registry):
+    """A raw xyd1 token names columns, never a figure: closed, no rebuild."""
+    app = make_plane_app()
+
+    async def main():
+        async with data_plane_server(rebuild=make_rebuild_hook(app)) as (url, _):
+            client = await connect_client(url)
+            collector = Collector(client)
+            await client.emit("sub", {"fig": plane_data_token(), "mid": "m1"}, namespace="/_xy")
+            error = await collector.next(collector.errors)
+            assert error["error"] == "unknown figure token"
+            await client.disconnect()
+
+    run(main())

From f9f617d0fb45d271b5d7c97620864c17d67583c2 Mon Sep 17 00:00:00 2001
From: Farhan Ali Raza <farhanalirazaazeemi@gmail.com>
Date: Thu, 6 Aug 2026 17:28:45 +0500
Subject: [PATCH 2/3] fix(reflex): unbind plan index on every unmount
 transition
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The data-token -> {digests} index was inserted on subscribe but only
pruned when a republish under the same session token later found the
plan unmounted — short-lived sessions accumulated bindings forever.
Every transition that can end a mount now funnels through
_unbind_plan_if_unmounted_locked: last unsubscribe, disconnect,
release, failed-rebuild cleanup, and the TTL sweep. A cached figure
entry keeps the binding alive (still mounted); dropping the last of
entry+subscribers drops it.
---
 python/reflex_xy/registry.py          | 44 +++++++++++++++++++++------
 spec/design/reflex-integration.md     | 10 ++++--
 tests/reflex_adapter/test_data_var.py | 39 ++++++++++++++++++++++++
 3 files changed, 82 insertions(+), 11 deletions(-)

diff --git a/python/reflex_xy/registry.py b/python/reflex_xy/registry.py
index 6ea57d8b..375b9e55 100644
--- a/python/reflex_xy/registry.py
+++ b/python/reflex_xy/registry.py
@@ -125,8 +125,10 @@ def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
         # 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.
+        # composite figure binds (namespace sub) and dropped by
+        # _unbind_plan_if_unmounted_locked on every transition that can end
+        # the mount — unsubscribe, disconnect, release, failed-rebuild
+        # cleanup, TTL sweep, and a republish that finds it unmounted.
         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
@@ -458,6 +460,7 @@ def release(self, token: str) -> None:
             self._invalidate_rebuild_guards_locked(token)
             entry = self._entries.pop(token, None)
             self._retain_removed_version_locked(token, entry)
+            self._unbind_plan_if_unmounted_locked(token)
 
     def remove_if_current(self, token: str, expected: FigureEntry, *, guard: object) -> bool:
         """Remove ``expected`` only while its rebuild guard is still valid.
@@ -474,11 +477,35 @@ def remove_if_current(self, token: str, expected: FigureEntry, *, guard: object)
                 return False
             del self._entries[token]
             self._retain_removed_version_locked(token, expected)
+            self._unbind_plan_if_unmounted_locked(token)
             return True
 
     def _invalidate_rebuild_guards_locked(self, token: str) -> None:
         self._active_rebuild_guards.pop(token, None)
 
+    def _unbind_plan_if_unmounted_locked(self, token: str) -> None:
+        """Drop a composite plan token's binding once nothing serves or
+        watches it (mutex held; no-op for non-plan tokens).
+
+        This is the other half of the index invariant "bounded by mounted
+        plans": ``bind_plan`` inserts on subscribe, and every transition that
+        can end a mount — unsubscribe, disconnect, release, failed-rebuild
+        cleanup, the TTL sweep — funnels here, so short-lived sessions cannot
+        accumulate bindings that only a republish would have collected.
+        """
+        from .tokens import parse_plan_token
+
+        parsed = parse_plan_token(token)
+        if parsed is None:
+            return
+        if token in self._entries or self._rebuildable_subscribers.get(token):
+            return
+        digests = self._digests_by_data_token.get(parsed.data_token)
+        if digests is not None:
+            digests.discard(parsed.digest)
+            if not digests:
+                self._digests_by_data_token.pop(parsed.data_token, None)
+
     def _retain_removed_version_locked(self, token: str, entry: Optional[FigureEntry]) -> None:
         if self._rebuildable_subscribers.get(token):
             version = 0 if entry is None else entry.version
@@ -518,6 +545,7 @@ def _unsubscribe_locked(self, token: str, sid: str) -> None:
             if not subscribers:
                 self._rebuildable_subscribers.pop(token, None)
                 self._evicted_versions.pop(token, None)
+                self._unbind_plan_if_unmounted_locked(token)
 
         tokens = self._rebuildable_tokens_by_sid.get(sid)
         if tokens is not None:
@@ -537,6 +565,7 @@ def disconnect(self, sid: str) -> None:
                 if not subscribers:
                     self._rebuildable_subscribers.pop(token, None)
                     self._evicted_versions.pop(token, None)
+                    self._unbind_plan_if_unmounted_locked(token)
 
     def tokens(self) -> list[str]:
         with self._mutex:
@@ -613,13 +642,9 @@ def _rebuild_dependent(self, data_token: str, digest: str, column_entry: ColumnE
             )
             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)
+                # binding (the same invariant every unmount transition
+                # enforces); a later subscribe rebuilds and re-indexes.
+                self._unbind_plan_if_unmounted_locked(composite)
                 return
         parsed = parse_token(data_token)
         source = (
@@ -897,6 +922,7 @@ def sweep(self, *, now: Optional[float] = None) -> list[str]:
                     self._invalidate_rebuild_guards_locked(token)
                     self._retain_removed_version_locked(token, entry)
                     del self._entries[token]
+                    self._unbind_plan_if_unmounted_locked(token)
                     dropped.append(token)
             for token, column_entry in list(self._column_entries.items()):
                 if now - column_entry.last_access > self._ttl:
diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md
index a3ed2187..882fad99 100644
--- a/spec/design/reflex-integration.md
+++ b/spec/design/reflex-integration.md
@@ -511,8 +511,14 @@ refused as a subscription outright.
 re-binds every mounted dependent through the `data token → {digests}`
 index — fresh figures, bumped versions, coalesced room broadcasts, exactly
 the figure-var republish machinery. The index is added to when a composite
-binds and pruned when a republish finds the plan unmounted, so it stays
-bounded by mounted plans. Failure stays loud: a bind that stops matching
+binds (namespace `sub`) and pruned on **every transition that can end the
+mount**: the last unsubscribe or disconnect for the composite token, its
+entry's release, failed-rebuild cleanup, the TTL sweep, and a republish
+that finds it unmounted (`_unbind_plan_if_unmounted_locked`). "Bounded by
+mounted plans" therefore holds for short-lived sessions too — cleanup never
+depends on that same session republishing later. Pinned by
+`test_data_var.py::test_unmount_transitions_drop_plan_bindings`. Failure
+stays loud: a bind that stops matching
 (possible only for untyped data vars) logs server-side, releases the
 composite entry, and answers the room `err {resync}`; a stale digest
 (hot-reload drift) answers `err {resync}` naming the digest.
diff --git a/tests/reflex_adapter/test_data_var.py b/tests/reflex_adapter/test_data_var.py
index 20698d4e..451177fb 100644
--- a/tests/reflex_adapter/test_data_var.py
+++ b/tests/reflex_adapter/test_data_var.py
@@ -229,6 +229,45 @@ def test_republish_drops_unmounted_dependents_from_the_index(_fresh_registry, cl
     assert _fresh_registry._digests_by_data_token.get(handle.token) is None
 
 
+def test_unmount_transitions_drop_plan_bindings(_fresh_registry, client_token):
+    """The plan index is bounded by *mounted* plans: unsubscribe/disconnect
+    and entry eviction (release, sweep) must unbind — never only a later
+    republish under the same session token (short-lived sessions would
+    otherwise accumulate bindings forever)."""
+    registry = _fresh_registry
+    data_token = f"xyd1|{client_token}|app.app.State|cloud"
+    digest = "ab" * 10
+    composite = build_plan_token(digest, data_token)
+
+    # subscriber-only mount (no cached figure): last unsubscribe unbinds
+    registry.bind_plan(data_token, digest)
+    registry.subscribe(composite, "sid-1", rebuildable=True)
+    registry.unsubscribe(composite, "sid-1")
+    assert registry._digests_by_data_token.get(data_token) is None
+
+    # same via disconnect
+    registry.bind_plan(data_token, digest)
+    registry.subscribe(composite, "sid-2", rebuildable=True)
+    registry.disconnect("sid-2")
+    assert registry._digests_by_data_token.get(data_token) is None
+
+    # a cached figure keeps the mount alive across unsubscribe...
+    registry.bind_plan(data_token, digest)
+    registry.subscribe(composite, "sid-3", rebuildable=True)
+    registry.publish(composite, xy.scatter_chart(xy.scatter([0.0], [0.0])).figure())
+    registry.unsubscribe(composite, "sid-3")
+    assert registry._digests_by_data_token.get(data_token) == {digest}
+    # ...until the entry itself goes: release unbinds
+    registry.release(composite)
+    assert registry._digests_by_data_token.get(data_token) is None
+
+    # and the TTL sweep unbinds an idle, unsubscribed composite entry
+    registry.bind_plan(data_token, digest)
+    entry = registry.publish(composite, xy.scatter_chart(xy.scatter([0.0], [0.0])).figure())
+    registry.sweep(now=entry.last_access + 10**9)
+    assert registry._digests_by_data_token.get(data_token) is None
+
+
 def test_release_columns_releases_dependent_figures(_fresh_registry, client_token):
     state = hydrated_substate(client_token)
     handle = state.maybe.token

From 0e39225f474e2fef7c00b3da0ca97c54fa5400f8 Mon Sep 17 00:00:00 2001
From: Masen Furer <m_github@0x26.net>
Date: Thu, 6 Aug 2026 22:01:38 +0000
Subject: [PATCH 3/3] fix(reflex): one bind-failure source label across both
 composite paths

The subscribe-time rebuild truncated the state's full name to its last
dotted segment while the republish fan-out used the full name, so one
mismatch produced two differently-labelled err frames (and the rebuilt
validate_columns label disagreed with both). Both state_bridge labels now
carry the full state name, matching registry._rebuild_dependent.
---
 python/reflex_xy/state_bridge.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/python/reflex_xy/state_bridge.py b/python/reflex_xy/state_bridge.py
index e76d03af..a451f93c 100644
--- a/python/reflex_xy/state_bridge.py
+++ b/python/reflex_xy/state_bridge.py
@@ -91,9 +91,7 @@ async def rebuild_data(app: Any, parsed: ParsedToken) -> Optional[dict[str, Any]
     columns = await _run_state_method(app, parsed)
     if columns is None:
         return None
-    return validate_columns(
-        columns, source=f"{parsed.state_full_name.rsplit('.', 1)[-1]}.{parsed.var_name}"
-    )
+    return validate_columns(columns, source=f"{parsed.state_full_name}.{parsed.var_name}")
 
 
 async def rebuild_plan_figure(app: Any, composite: ParsedPlanToken) -> Optional["Figure"]:
@@ -111,7 +109,9 @@ async def rebuild_plan_figure(app: Any, composite: ParsedPlanToken) -> Optional[
         # Future binds (other plans over the same data var) hit the cache;
         # bind_plan below is the subscribe path's job, not the rebuild's.
         registry.publish_columns(composite.data_token, columns)
-    source = f"{composite.data.state_full_name.rsplit('.', 1)[-1]}.{composite.data.var_name}"
+    # Same label the registry's republish path builds in _rebuild_dependent:
+    # one mismatch, one err frame, whichever path hits it first.
+    source = f"{composite.data.state_full_name}.{composite.data.var_name}"
     return plan.bind(columns, source=source).figure()