Avoid copies in live update attachments

CHANGELOG.md

@@ -196,6 +196,12 @@ in the README).
   assembly retains encoded arrays until the final blob join instead of copying
   every column through `tobytes()` first. Payload bytes and sampling decisions
   remain parity-tested and unchanged.
+- **Zero-copy live-update attachments.** Re-decimation, density/drill, and
+  selection replies now expose owner-retaining contiguous byte `memoryview`s
+  instead of copying every encoded ndarray through `tobytes()`. Anywidget and
+  scatter/gather frame transports borrow the views directly; the Reflex
+  socket.io adapter keeps its one explicit conversion at the owned-bytes wire
+  boundary. Frame bytes and selection masks remain byte-for-byte unchanged.
 - **Stable hybrid density overlays.** Pyramid-served pan/zoom updates now keep
   the retained deterministic point sample when they omit a replacement,
   instead of making the first-paint overlay disappear on interaction. Exact

python/reflex-xy/reflex_xy/namespace.py

@@ -38,6 +38,7 @@
 
 from socketio import AsyncNamespace
 
+from xy._buffers import WireBuffer
 from xy.channel import handle_message
 
 from .registry import FigureEntry, FigureRegistry
@@ -185,7 +186,7 @@ async def on_msg(self, sid: str, data: Any) -> None:
     # -- server-side pushes (append/refresh fan-out) ---------------------------
 
     async def broadcast_message(
-        self, token: str, message: dict[str, Any], buffers: Optional[list[bytes]] = None
+        self, token: str, message: dict[str, Any], buffers: Optional[list[WireBuffer]] = None
     ) -> None:
         """Push one channel message to every subscriber of a figure."""
         await self.emit(

python/reflex-xy/reflex_xy/registry.py

@@ -25,6 +25,7 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
+    from xy._buffers import WireBuffer
     from xy._figure import Figure
 
 __all__ = ["FigureEntry", "FigureRegistry", "registry"]
@@ -77,7 +78,7 @@ def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
         self._on_publish: Optional[Callable[[str, FigureEntry], Awaitable[None]]] = None
         # async callback(token, message, buffers) -> None for incremental
         # pushes (append) — same seam, message-shaped instead of payload-shaped.
-        self._on_push: Optional[Callable[[str, dict, list[bytes]], Awaitable[None]]] = None
+        self._on_push: Optional[Callable[[str, dict, list[WireBuffer]], Awaitable[None]]] = None
 
     # -- wiring ------------------------------------------------------------
 
@@ -87,7 +88,7 @@ def attach_loop(self, loop: asyncio.AbstractEventLoop) -> None:
     def on_publish(self, callback: Callable[[str, FigureEntry], Awaitable[None]]) -> None:
         self._on_publish = callback
 
-    def on_push(self, callback: Callable[[str, dict, list[bytes]], Awaitable[None]]) -> None:
+    def on_push(self, callback: Callable[[str, dict, list[WireBuffer]], Awaitable[None]]) -> None:
         self._on_push = callback
 
     # -- core map ----------------------------------------------------------
@@ -244,7 +245,7 @@ async def _do() -> None:
             asyncio.run_coroutine_threadsafe(_do(), loop)
 
     def push_view_message(
-        self, token: str, build: Callable[["Figure"], tuple[dict, list[bytes]]]
+        self, token: str, build: Callable[["Figure"], tuple[dict, list[WireBuffer]]]
     ) -> None:
         """Build one kernel→client message against a figure and push it
         room-wide (spec/design/view-state.md §5.2).
@@ -312,7 +313,7 @@ def select(
             self.push_view_message(token, lambda fig: fig.selection_rows_message(rows))
             return
 
-        def build(fig: "Figure") -> tuple[dict, list[bytes]]:
+        def build(fig: "Figure") -> tuple[dict, list[WireBuffer]]:
             selection = fig._validated_state_selection(range=range, polygon=polygon)
             if selection is None:
                 msg = "select() needs range=, polygon=, or rows="

python/xy/_buffers.py

@@ -0,0 +1,27 @@
+"""Internal binary-attachment types and zero-copy NumPy byte views.
+
+Live update transports accept buffer-protocol objects, so an encoded ndarray
+does not need to become an owned ``bytes`` object before it reaches the actual
+wire boundary. A ``memoryview`` retains its ndarray owner for as long as the
+transport needs the attachment.
+"""
+
+from __future__ import annotations
+
+from typing import TypeAlias
+
+import numpy as np
+
+WireBuffer: TypeAlias = bytes | memoryview
+
+
+def array_byte_view(array: np.ndarray) -> memoryview:
+    """Return an owner-retaining byte view over a C-contiguous ndarray.
+
+    Non-contiguous inputs are copied only to satisfy the wire's contiguous
+    buffer contract. Dtype conversion, when required, stays explicit at the
+    caller so the encoded scalar format remains visible there.
+    """
+
+    owner = np.ascontiguousarray(array)
+    return owner.data.cast("B")

python/xy/_figure.py

@@ -18,6 +18,7 @@
 from . import _annotations, _validate, channels, columns, export, interaction, kernels, styles
 from . import marks as _marks
 from ._annotations import AnnotationsMixin
+from ._buffers import WireBuffer, array_byte_view
 from ._payload import PayloadMixin
 from ._trace import Trace
 from .channels import ColorChannel, SizeChannel
@@ -1347,7 +1348,7 @@ def _rect_finite_sel(
 
     def density_view(
         self, trace_id: int, x0: float, x1: float, y0: float, y1: float, w: int, h: int
-    ) -> tuple[dict[str, Any], list[bytes]]:
+    ) -> tuple[dict[str, Any], list[WireBuffer]]:
         """Re-bin a density-mode scatter's aggregation grid for a new viewport."""
         return interaction.density_view(self, trace_id, x0, x1, y0, y1, w, h)
 
@@ -1376,7 +1377,7 @@ def to_shipped_indices(self, trace_id: int, canonical: np.ndarray) -> np.ndarray
 
     def decimate_view(
         self, x0: float, x1: float, px_width: int
-    ) -> tuple[dict[str, Any], list[bytes]]:
+    ) -> tuple[dict[str, Any], list[WireBuffer]]:
         """Re-decimate the visible line windows on zoom, re-centering the
         f32 upload offsets so precision holds at deep zoom."""
         return interaction.decimate_view(self, x0, x1, px_width)
@@ -1394,7 +1395,7 @@ def append(
         alpha: Any = None,
         stroke_width: Any = None,
         symbol: Any = None,
-    ) -> tuple[dict[str, Any], list[bytes]]:
+    ) -> tuple[dict[str, Any], list[WireBuffer]]:
         """Streaming append: extend a scatter/line trace's canonical columns
         and get the client refresh message back. The widget's `append` sends
         it; headless callers can inspect or discard it. Payloads stay
@@ -1506,7 +1507,7 @@ def view_nav_message(self, axes: Any = None) -> dict[str, Any]:
             message["axes"] = self._axis_policy(tuple(axes), "reset axes")
         return message
 
-    def selection_rows_message(self, rows: Any) -> tuple[dict[str, Any], list[bytes]]:
+    def selection_rows_message(self, rows: Any) -> tuple[dict[str, Any], list[WireBuffer]]:
         """Kernel-resolve a per-trace row-index selection into the same binary
         mask buffers the gesture selection path ships (§5.1). Rows-selections
         are non-durable by design; the client applies them outside history."""
@@ -1515,7 +1516,7 @@ def selection_rows_message(self, rows: Any) -> tuple[dict[str, Any], list[bytes]
         if not isinstance(rows, dict):
             rows = {0: rows}
         traces: list[dict[str, Any]] = []
-        out: list[bytes] = []
+        out: list[WireBuffer] = []
         total = 0
         for trace_id, indices in rows.items():
             tid = int(trace_id)
@@ -1557,7 +1558,7 @@ def selection_rows_message(self, rows: Any) -> tuple[dict[str, Any], list[bytes]
                     "drill_seq": self.traces[tid].drill_seq,
                 }
             )
-            out.append(wire_idx.tobytes())
+            out.append(array_byte_view(wire_idx))
             # Deduplicated, validated canonical rows — not the raw request
             # length and not only the currently-shipped subset.
             total += int(idx.size)

python/xy/channel.py

@@ -32,10 +32,11 @@
 from __future__ import annotations
 
 import math
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Optional
 
+from ._buffers import WireBuffer, array_byte_view
 from ._figure import Selection
 from ._framing import (
     DEFAULT_FRAME_LIMITS,
@@ -79,7 +80,7 @@
 ]
 
 # (reply message, buffers to ship beside it — None when the reply has none).
-Reply = tuple[dict[str, Any], Optional[list[bytes]]]
+Reply = tuple[dict[str, Any], Optional[list[WireBuffer]]]
 
 # Reflex semantic selection events include bounded JSON projections. The
 # complete canonical Selection remains server-side and can be re-resolved.
@@ -120,7 +121,7 @@ def _selection_reply(
     if callbacks.on_brush is not None:
         callbacks.on_brush(brush)
     traces = []
-    out: list[bytes] = []
+    out: list[WireBuffer] = []
     total = 0
     for tid, idx in selected.items():
         # The wire mask speaks shipped-vertex positions; callbacks retain
@@ -134,10 +135,16 @@ def _selection_reply(
                 "drill_seq": fig.traces[tid].drill_seq,
             }
         )
-        out.append(wire_idx.tobytes())
+        out.append(array_byte_view(wire_idx))
         total += len(idx)
     if callbacks.on_select is not None:
-        callbacks.on_select(Selection(fig, selected))
+        # The outgoing memoryviews may borrow `selected` directly when the
+        # canonical and shipped index spaces are identical. Give user code an
+        # isolated callback snapshot so mutation cannot rewrite an attachment
+        # after it was assembled; callback-free transports keep the zero-copy
+        # path and pay no canonical-index copy.
+        callback_selected = {tid: idx.copy() for tid, idx in selected.items()}
+        callbacks.on_select(Selection(fig, callback_selected))
     message: dict[str, Any] = {"type": "selection", "traces": traces, "total": total}
     if seq is not None:
         message["seq"] = seq
@@ -181,7 +188,7 @@ def _selection_reply(
 def handle_message(
     fig: "Figure",
     content: Any,
-    buffers: Optional[list[bytes]] = None,
+    buffers: Optional[Sequence[WireBuffer]] = None,
     callbacks: ChannelCallbacks = _NO_CALLBACKS,
 ) -> Optional[Reply]:
     """Dispatch one client message against a figure.

python/xy/interaction.py

@@ -20,6 +20,7 @@
 import numpy as np
 
 from . import channels, columns, kernels, lod
+from ._buffers import WireBuffer, array_byte_view
 from .config import (
     DECIMATION_THRESHOLD,
     DEFAULT_PALETTE,
@@ -307,7 +308,7 @@ def _point_shipped_sel(t: Any) -> Optional[np.ndarray]:
 
 def decimate_view(
     fig: "Figure", x0: float, x1: float, px_width: int
-) -> tuple[dict[str, Any], list[bytes]]:
+) -> tuple[dict[str, Any], list[WireBuffer]]:
     """Re-decimate visible windows for a zoomed view (recompute for the
     visible x-range only; design dossier §28). The offset re-centers on the
     window midpoint — the §16 deep-zoom rule — so f32 precision follows the
@@ -410,15 +411,15 @@ def pyramid_report_bytes(fig: Any) -> int:
     return sum(_pyramid_resident_bytes() for t in fig.traces if getattr(t, "_pyr_handle", 0))
 
 
-def _encode_log_u8(grid: np.ndarray) -> tuple[bytes, float]:
-    """Density grid -> log-encoded u8 wire bytes (client decodes via expm1).
+def _encode_log_u8(grid: np.ndarray) -> tuple[memoryview, float]:
+    """Density grid -> log-encoded u8 wire view (client decodes via expm1).
     Zero cells stay zero; any nonzero cell maps to at least 1 so the "lit if
     occupied" texture contract survives quantization."""
     enc, maximum = kernels.density_log_u8(np.asarray(grid, dtype=np.float32))
-    return enc.tobytes(), maximum
+    return array_byte_view(enc), maximum
 
 
-def _decode_log_u8(buf: bytes, gmax: float) -> np.ndarray:
+def _decode_log_u8(buf: WireBuffer, gmax: float) -> np.ndarray:
     """Inverse of :func:`_encode_log_u8` — the Python twin of the client's
     ``lodDecodeLogU8`` and the executable wire contract for tests. Lossy
     (8-bit in log space, sub-percent per-cell at typical maxima), but zeros
@@ -489,7 +490,7 @@ def _density_sample_update(
 
 def density_view(
     fig: "Figure", trace_id: int, x0: float, x1: float, y0: float, y1: float, w: int, h: int
-) -> tuple[dict[str, Any], list[bytes]]:
+) -> tuple[dict[str, Any], list[WireBuffer]]:
     """Re-aggregate a density-mode scatter for a new viewport (O(visible
     points); the client requests this when pan/zoom leaves the shipped grid).
 
@@ -602,7 +603,7 @@ def _drill_points(
     hi_y: float,
     w: int,
     h: int,
-) -> tuple[dict[str, Any], list[bytes]]:
+) -> tuple[dict[str, Any], list[WireBuffer]]:
     """Ship the visible subset of a Tier-2 scatter as real points (§5 drill-in).
 
     Scatter-specific wiring over the chart-agnostic pieces in `lod`: channels
@@ -675,7 +676,7 @@ def append_data(
     alpha: Any = None,
     stroke_width: Any = None,
     symbol: Any = None,
-) -> tuple[dict[str, Any], list[bytes]]:
+) -> tuple[dict[str, Any], list[WireBuffer]]:
     """Streaming append (Phase-0): extend a trace's canonical
     columns in place and return the client refresh message.
 

python/xy/lod.py

@@ -27,6 +27,7 @@
 import numpy as np
 
 from . import kernels
+from ._buffers import WireBuffer, array_byte_view
 from .config import DENSITY_TARGET_POINTS_PER_CELL, DRILL_EXIT_FACTOR, MAX_SCREEN_DIM
 
 _SPLITMIX_INCREMENT = np.uint64(0x9E3779B97F4A7C15)
@@ -739,19 +740,21 @@ class BufferWriter:
     every tiered chart's incremental updates use."""
 
     def __init__(self) -> None:
-        self.buffers: list[bytes] = []
+        self.buffers: list[WireBuffer] = []
 
     def add_f32(self, arr: np.ndarray) -> int:
         """Append ``arr`` as a contiguous f32 buffer; returns its index."""
-        self.buffers.append(np.ascontiguousarray(arr, dtype=np.float32).tobytes())
+        encoded = np.ascontiguousarray(arr, dtype=np.float32)
+        self.buffers.append(array_byte_view(encoded))
         return len(self.buffers) - 1
 
     def add_u8(self, arr: np.ndarray) -> int:
         """Append ``arr`` as a flat u8 buffer; returns its index."""
-        self.buffers.append(np.ascontiguousarray(arr, dtype=np.uint8).reshape(-1).tobytes())
+        encoded = np.ascontiguousarray(arr, dtype=np.uint8).reshape(-1)
+        self.buffers.append(array_byte_view(encoded))
         return len(self.buffers) - 1
 
-    def add_raw(self, raw: bytes) -> int:
+    def add_raw(self, raw: WireBuffer) -> int:
         """Append pre-encoded bytes untouched; returns their index."""
         self.buffers.append(raw)
         return len(self.buffers) - 1

spec/design-dossier.md

@@ -931,6 +931,12 @@ number-parse-shaped; (c) every binding reports its actual copy count at ingest i
 mode, so "zero-copy" regressions are observable rather than folklore; (d) the Jupyter
 live path still has no text encoding of numbers, DOM payload, or main-thread data parse.
 
+For live view updates, the encoded contiguous ndarray is itself the attachment owner:
+Python exposes an owner-retaining byte `memoryview` rather than making an additional
+`ndarray.tobytes()` copy. Anywidget comms and scatter/gather `XYBF` framing consume the
+view directly. A binding whose protocol requires owned bytes (currently python-socketio)
+performs its single explicit conversion at that final transport boundary.
+
 ## 30. Compatibility subset — v1 is a list, not an aspiration
 
 Full Plotly semantics (~40 trace types × transforms × axis quirks × hover rules) is a

spec/design/lod-architecture.md

@@ -70,7 +70,12 @@ Tiered chart kinds must enter through the common LOD primitives in
   `encode_window_xy_columns(...)`, `add_window_xy(...)`, and
   `BufferWriter.add_encoded(...)` are the shared geometry wire primitive: f64
   data-space values become finite f32 buffers plus `{offset, scale, len}`
-  metadata in exactly one place.
+  metadata in exactly one place. `BufferWriter` retains each contiguous
+  encoded ndarray through an owner-holding byte `memoryview`; update assembly
+  never follows that required encoding allocation with `ndarray.tobytes()`.
+  The same attachment rule covers log-u8 density grids and u32 selection
+  masks; only transports that require owned bytes may convert at their final
+  boundary.
 - `sample_rows_for_target(...)` is the shared target-bounded subset primitive:
   density overlays and future sampled tiers ask for "about N stable rows from
   this viewport" in one place instead of copying target-fraction math.

spec/design/reflex-integration.md

@@ -109,7 +109,9 @@ namespace socket re-CONNECTs automatically, and every mounted chart re-`sub`s
 on the `connect` event.
 
 **Wire shape.** Metadata is one small JSON object per event; every data
-column is a `bytes` value inside it, which python-socketio hoists into
+column is converted from the kernel's owner-retaining `memoryview` to a
+`bytes` value at `_buffer_bytes` (the one boundary copy required by
+python-socketio), which python-socketio hoists into
 binary attachments and the browser receives as `ArrayBuffer`s *in place* —
 aligned, zero-copy into `Float32Array`s. No JSON numbers for data, no
 base64, no custom framing (§29 preserved; the socket.io protocol already

spec/design/wire-protocol.md

@@ -19,6 +19,16 @@ emits it (§2).
 returns either `None` or `(message, buffers)`, where `buffers` is a list of
 binary attachments the reply's spec entries index into by position.
 
+Kernel-produced attachments use the internal `WireBuffer = bytes | memoryview`
+type. View-dependent f32/u8 columns and u32 selection masks are C-contiguous
+byte memoryviews that retain their encoded ndarray owner; their bytes are
+therefore backed by valid storage for the attachment lifetime without a second
+`tobytes()` allocation. Senders must not mutate an attachment until its
+transport has consumed it. Anywidget comms and `encode_frame_parts` consume
+those views directly. A transport that requires owned bytes converts at its
+own boundary: the socket.io adapter does so once in
+`reflex_xy.namespace._buffer_bytes`.
+
 - Non-dict `content`, an unknown `type`, a missing required field, or a value
   that fails coercion returns `None`. Client-supplied data never raises;
   exceptions from *user callbacks* do propagate.
@@ -148,6 +158,12 @@ while `total` sums the canonical index counts; Python callbacks receive a
 assembled and `on_select` after — that order is the invariant. An empty
 `traces` list means "clear", and carries no buffers.
 
+All update attachments above are byte-for-byte identical to their owned-bytes
+encoding. If an input array is already C-contiguous and has the wire dtype, the
+view shares it; a non-contiguous or wrong-dtype input first materializes the
+single required contiguous encoded array, then borrows that array without a
+second payload-sized copy.
+
 ## 4. Server push
 
 **`append`** — `{type: "append", affected: [trace_id], spec}` with a single

tests/reflex_adapter/test_socket_data_plane.py

@@ -173,6 +173,7 @@ async def main():
             assert sel["message"]["type"] == "selection"
             assert sel["message"]["total"] == 8
             assert len(sel["buffers"]) == 1
+            assert isinstance(sel["buffers"][0], (bytes, bytearray))
 
             # malformed messages are dropped silently, never crash the server
             await client.emit("msg", {"fig": token, "m": ["not", "a", "dict"]}, namespace="/_xy")