From 6527b06d17ccb80de5aa1acf041fb4ffd0c4ab8b Mon Sep 17 00:00:00 2001
From: Alek Petuskey <alek@reflex.dev>
Date: Wed, 22 Jul 2026 00:28:50 -0700
Subject: [PATCH] Avoid copies in live update attachments

---
 CHANGELOG.md                                  |  6 ++
 python/reflex-xy/reflex_xy/namespace.py       |  3 +-
 python/reflex-xy/reflex_xy/registry.py        |  9 +--
 python/xy/_buffers.py                         | 27 ++++++++
 python/xy/_figure.py                          | 13 ++--
 python/xy/channel.py                          | 19 ++++--
 python/xy/interaction.py                      | 17 ++---
 python/xy/lod.py                              | 11 ++--
 spec/design-dossier.md                        |  6 ++
 spec/design/lod-architecture.md               |  7 +-
 spec/design/reflex-integration.md             |  4 +-
 spec/design/wire-protocol.md                  | 16 +++++
 .../reflex_adapter/test_socket_data_plane.py  |  1 +
 tests/test_channel.py                         | 38 +++++++++++
 tests/test_lod.py                             | 66 +++++++++++++++++++
 tests/test_scatter.py                         |  2 +
 tests/test_view_state.py                      |  9 ++-
 tests/test_widget.py                          |  1 +
 18 files changed, 222 insertions(+), 33 deletions(-)
 create mode 100644 python/xy/_buffers.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index da223dcb..d08b9e8d 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/python/reflex-xy/reflex_xy/namespace.py b/python/reflex-xy/reflex_xy/namespace.py
index 5fa7136c..0b804b29 100644
--- a/python/reflex-xy/reflex_xy/namespace.py
+++ b/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(
diff --git a/python/reflex-xy/reflex_xy/registry.py b/python/reflex-xy/reflex_xy/registry.py
index 00e2a2b8..a52e8cd3 100644
--- a/python/reflex-xy/reflex_xy/registry.py
+++ b/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="
diff --git a/python/xy/_buffers.py b/python/xy/_buffers.py
new file mode 100644
index 00000000..65a46e0e
--- /dev/null
+++ b/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")
diff --git a/python/xy/_figure.py b/python/xy/_figure.py
index 5493d5e8..ac2a90ab 100644
--- a/python/xy/_figure.py
+++ 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)
diff --git a/python/xy/channel.py b/python/xy/channel.py
index a50a0f34..ec9f0fd5 100644
--- a/python/xy/channel.py
+++ b/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.
diff --git a/python/xy/interaction.py b/python/xy/interaction.py
index b8ad655b..5eeb8c5d 100644
--- a/python/xy/interaction.py
+++ b/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.
 
diff --git a/python/xy/lod.py b/python/xy/lod.py
index fec719f7..03601cb4 100644
--- a/python/xy/lod.py
+++ b/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
diff --git a/spec/design-dossier.md b/spec/design-dossier.md
index 9df9427e..3cc01ce6 100644
--- a/spec/design-dossier.md
+++ b/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
diff --git a/spec/design/lod-architecture.md b/spec/design/lod-architecture.md
index 9691c1a1..62f94194 100644
--- a/spec/design/lod-architecture.md
+++ b/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.
diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md
index 95211424..b0a541ae 100644
--- a/spec/design/reflex-integration.md
+++ b/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
diff --git a/spec/design/wire-protocol.md b/spec/design/wire-protocol.md
index 60a7df51..5117e2a1 100644
--- a/spec/design/wire-protocol.md
+++ b/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
diff --git a/tests/reflex_adapter/test_socket_data_plane.py b/tests/reflex_adapter/test_socket_data_plane.py
index 8a00d058..fff74f14 100644
--- a/tests/reflex_adapter/test_socket_data_plane.py
+++ b/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")
diff --git a/tests/test_channel.py b/tests/test_channel.py
index d7a7e2dc..76dd7e9f 100644
--- a/tests/test_channel.py
+++ b/tests/test_channel.py
@@ -283,6 +283,43 @@ def test_select_fires_brush_before_select_and_returns_selection_reply():
     assert msg["seq"] == "selection:7"
     assert msg["total"] == 4
     assert buffers is not None and len(buffers) == len(msg["traces"])
+    assert all(isinstance(buffer, memoryview) for buffer in buffers)
+    assert all(isinstance(buffer.obj, np.ndarray) for buffer in buffers)
+
+
+def test_select_callback_mutation_cannot_rewrite_outgoing_attachment() -> None:
+    fig = Figure().scatter(np.arange(10.0), np.arange(10.0))
+    callback_rows = []
+
+    def mutate_selection(selection: Selection) -> None:
+        callback_rows.append(selection.index.copy())
+        selection.per_trace[0][:] = 9
+
+    reply = handle(
+        fig,
+        {"type": "select", "x0": 2.0, "x1": 5.0, "y0": 0.0, "y1": 6.0},
+        on_select=mutate_selection,
+    )
+
+    assert reply is not None
+    message, buffers = reply
+    assert message["total"] == 4
+    assert buffers is not None
+    np.testing.assert_array_equal(callback_rows[0], [2, 3, 4, 5])
+    np.testing.assert_array_equal(np.frombuffer(buffers[0], dtype=np.uint32), [2, 3, 4, 5])
+
+
+def test_select_without_callback_borrows_mask_owner(monkeypatch) -> None:
+    fig = Figure().scatter(np.arange(10.0), np.arange(10.0))
+    canonical = np.asarray([2, 3, 4, 5], dtype=np.uint32)
+    monkeypatch.setattr(fig, "select_range", lambda *_args: {0: canonical})
+
+    reply = handle(fig, {"type": "select", "x0": 2.0, "x1": 5.0, "y0": 0.0, "y1": 6.0})
+
+    assert reply is not None
+    _message, buffers = reply
+    assert buffers is not None
+    assert buffers[0].obj is canonical
 
 
 def test_lasso_select_returns_only_points_inside_polygon():
@@ -338,6 +375,7 @@ def test_select_wire_mask_is_shipped_space_selection_is_canonical():
     # ship time, so canonical rows [0, 2, 4] became shipped [0, 1, 2].
     wire = np.frombuffer(buffers[0], dtype=np.uint32)
     np.testing.assert_array_equal(wire, [1, 2])
+    assert isinstance(buffers[0], memoryview)
     assert msg["traces"][0]["count"] == 2
 
 
diff --git a/tests/test_lod.py b/tests/test_lod.py
index e8b641c4..7a49266f 100644
--- a/tests/test_lod.py
+++ b/tests/test_lod.py
@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import gc
+import weakref
 from pathlib import Path
 
 import numpy as np
 import pytest
 
 from xy import lod
+from xy.channel import encode_frame, encode_frame_parts
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -258,6 +261,69 @@ def test_encode_f32_values_and_buffer_writer_share_wire_contract() -> None:
     assert len(writer.buffers) == 1
 
 
+def test_buffer_writer_views_retain_owner_and_share_matching_input() -> None:
+    source = np.arange(8, dtype=np.float32)
+    source_ref = weakref.ref(source)
+    writer = lod.BufferWriter()
+
+    assert writer.add_f32(source) == 0
+    view = writer.buffers[0]
+
+    assert isinstance(view, memoryview)
+    assert view.format == "B" and view.c_contiguous
+    assert view.obj is source
+    source[3] = 42.5
+    assert np.frombuffer(view, dtype=np.float32)[3] == 42.5
+
+    # The attachment itself owns the ndarray lifetime; callers do not need to
+    # keep either their source name or the writer alive until transport send.
+    del source, writer
+    gc.collect()
+    assert source_ref() is view.obj
+    assert np.frombuffer(view, dtype=np.float32)[3] == 42.5
+    del view
+    gc.collect()
+    assert source_ref() is None
+
+
+def test_buffer_writer_copies_only_required_noncontiguous_conversions() -> None:
+    f32_base = np.arange(24, dtype=np.float64)
+    f32_source = f32_base[::3]
+    u8_base = np.arange(20, dtype=np.uint8)
+    u8_source = u8_base[::2]
+    expected_f32 = np.ascontiguousarray(f32_source, dtype=np.float32).tobytes()
+    expected_u8 = np.ascontiguousarray(u8_source, dtype=np.uint8).reshape(-1).tobytes()
+    writer = lod.BufferWriter()
+
+    writer.add_f32(f32_source)
+    writer.add_u8(u8_source)
+
+    assert all(isinstance(buffer, memoryview) for buffer in writer.buffers)
+    assert all(
+        buffer.c_contiguous and isinstance(buffer.obj, np.ndarray) for buffer in writer.buffers
+    )
+    assert [bytes(buffer) for buffer in writer.buffers] == [expected_f32, expected_u8]
+    f32_base[:] = -1.0
+    u8_base[:] = 255
+    assert [bytes(buffer) for buffer in writer.buffers] == [expected_f32, expected_u8]
+
+
+def test_update_views_preserve_exact_frame_bytes_without_payload_copy() -> None:
+    writer = lod.BufferWriter()
+    writer.add_f32(np.arange(30, dtype=np.float64).reshape(5, 6)[:, ::2])
+    writer.add_u8(np.arange(24, dtype=np.uint8).reshape(4, 6)[:, ::2])
+    owned = [bytes(buffer) for buffer in writer.buffers]
+    message = {"type": "tier_update", "seq": 9, "traces": [{"id": 0}]}
+
+    assert encode_frame(message, writer.buffers) == encode_frame(message, owned)
+    parts = encode_frame_parts(message, writer.buffers)
+    buffer_parts = [part for part in parts if isinstance(part, memoryview)]
+    assert [bytes(part) for part in buffer_parts] == owned
+    assert all(
+        part.obj is buffer.obj for part, buffer in zip(buffer_parts, writer.buffers, strict=True)
+    )
+
+
 def test_add_window_xy_uses_single_shared_buffer_writer_contract() -> None:
     xs = np.array([9.0, 10.0, 11.0], dtype=np.float64)
     ys = np.array([98.0, 100.0, 102.0], dtype=np.float64)
diff --git a/tests/test_scatter.py b/tests/test_scatter.py
index acceb850..c544ada0 100644
--- a/tests/test_scatter.py
+++ b/tests/test_scatter.py
@@ -597,6 +597,8 @@ def test_density_view_rebins():
     # cell, with `max` restoring the scale on decode.
     assert d["enc"] == "log-u8"
     assert len(buffers[0]) == 64 * 48
+    assert isinstance(buffers[0], memoryview)
+    assert isinstance(buffers[0].obj, np.ndarray)
     grid = _decode_log_u8(buffers[0], d["max"])
     assert len(grid) == 64 * 48
     assert grid.max() == pytest.approx(d["max"])  # grid max survives exactly
diff --git a/tests/test_view_state.py b/tests/test_view_state.py
index e135d973..48bdc566 100644
--- a/tests/test_view_state.py
+++ b/tests/test_view_state.py
@@ -132,13 +132,18 @@ def test_selection_rows_message_ships_mask_buffers() -> None:
     # Same buffers the gesture selection path ships: shipped vertex indices.
     expected = fig.to_shipped_indices(0, np.asarray([1, 3, 5]))
     assert buffers[0] == expected.tobytes()
+    assert isinstance(buffers[0], memoryview)
+    assert isinstance(buffers[0].obj, np.ndarray)
 
 
 def test_selection_rows_message_bare_array_is_trace_zero() -> None:
     fig = _figure()
-    msg, _buffers = fig.selection_rows_message([2, 4])
+    source = np.arange(8, dtype=np.uint32)[::2]
+    msg, buffers = fig.selection_rows_message(source)
     assert msg["traces"][0]["id"] == 0
-    assert msg["total"] == 2
+    assert msg["total"] == 4
+    np.testing.assert_array_equal(np.frombuffer(buffers[0], dtype=np.uint32), source)
+    assert isinstance(buffers[0], memoryview) and buffers[0].c_contiguous
 
 
 def test_selection_rows_message_rejects_unknown_trace() -> None:
diff --git a/tests/test_widget.py b/tests/test_widget.py
index b34715ac..122df87e 100644
--- a/tests/test_widget.py
+++ b/tests/test_widget.py
@@ -140,3 +140,4 @@ def test_widget_emits_brush_range_before_selection_callback():
     assert content["type"] == "selection"
     assert content["total"] == 4
     assert buffers is not None
+    assert all(isinstance(buffer, memoryview) for buffer in buffers)