Ship streaming append once, split, per tick (protocol v5)

CHANGELOG.md

@@ -142,6 +142,14 @@ in the README).
   contract without importing the widget stack.
 
 ### Changed
+- **Streaming append ships once, split, per tick (protocol v5).** The
+  `append` refresh now uses the same split buffer layout as first paint (no
+  packed join copy), and on the notebook widget it rides the single
+  `spec`/`buffers` trait update — which doubles as reopen state — instead of
+  being transmitted twice (trait re-sync plus a custom message). The client
+  applies appends when `spec.append.seq` advances; the Reflex socket push is
+  unchanged in shape apart from the split buffers. Halves streaming wire
+  bytes and removes two full-payload copies per tick.
 - **Responsive, author-defeatable browser chrome.** XY's visual defaults now
   live in a low-priority cascade layer, so Tailwind utilities, ordinary author
   CSS, and slot styles override them without `!important`. Long legends remain

js/src/00_header.ts

@@ -20,7 +20,7 @@
  * title, axis tick labels, legend, tooltip (§7).
  */
 
-export const PROTOCOL = 4;
+export const PROTOCOL = 5;
 
 // HTTP binary frame v1 (spec/design/wire-protocol.md §7; Python side in
 // python/xy/_framing.py). The chart spec's PROTOCOL
@@ -53,6 +53,47 @@ export function bytesToSpan(b) {
   return span.byteOffset % 4 === 0 ? span : new Uint8Array(span);
 }
 
+/** True when every column in the spec fits inside the supplied buffers.
+ * Trait-transported appends can observe a torn update on hosts that set
+ * spec and buffers non-atomically (one change event fires between the two
+ * writes); a torn pair must be deferred to the other change event, not
+ * applied and not treated as fatal. Message- and first-paint transports
+ * deliver the pair together, so for them a mismatch stays a loud
+ * `payloadBuffers`/`_columnView` error, never this soft check. */
+export function payloadCoherent(spec, raw) {
+  const cols = spec?.columns;
+  if (!Array.isArray(cols)) return false;
+  const size = (c) => (c.dtype === "u8" ? 1 : 4);
+  if (spec.buffer_layout === "split") {
+    if (!Array.isArray(raw)) return false;
+    return cols.every((c) => {
+      if (!Number.isInteger(c.buf)) return true; // raster-only borrowed span
+      const b = raw[c.buf];
+      return !!b && c.len * size(c) <= b.byteLength;
+    });
+  }
+  if (Array.isArray(raw) || !raw) return false;
+  return cols.every((c) => c.byte_offset + c.len * size(c) <= raw.byteLength);
+}
+
+/** Wire buffers in the shape the spec declares (§29): packed is one blob;
+ * split is one span per column. Aligned views stay zero-copy; only a
+ * legacy unaligned view pays a narrow view-sized copy. A spec/transport
+ * disagreement is a bug, never a fallback. Used at first paint and by the
+ * streaming-append apply path (both ride the same layouts). */
+export function payloadBuffers(spec, raw) {
+  if (spec.buffer_layout === "split") {
+    if (!Array.isArray(raw)) {
+      throw new Error("xy: spec says buffer_layout=split but the transport delivered one buffer");
+    }
+    return raw.map(bytesToSpan);
+  }
+  if (Array.isArray(raw)) {
+    throw new Error("xy: transport delivered a buffer list but the spec is not split-layout");
+  }
+  return bytesToSpan(raw);
+}
+
 function xyFrameLimit(limits, name) {
   const fallback = XY_FRAME_DEFAULT_LIMITS[name];
   const value = limits && limits[name] != null ? limits[name] : fallback;

js/src/54_kernel.ts

@@ -1,4 +1,4 @@
-import { bytesToSpan } from "./00_header";
+import { payloadBuffers } from "./00_header";
 import { lodApplyDensityUpdate, lodApplyDrill, lodDropDrill, lodRememberDensity } from "./45_lod";
 import { xyCreateRebinWorker } from "./46_worker";
 import { ChartView } from "./50_chartview";
@@ -197,12 +197,15 @@ Object.assign(ChartView.prototype, {
   // matter how much data has accumulated — and names the traces whose data
   // changed. Only those GPU traces rebuild; everything else keeps its state.
   // Tiered traces then refine to the *current* window through the normal
-  // stale-while-revalidate request path (§17).
+  // stale-while-revalidate request path (§17). The payload arrives in
+  // whichever layout the spec declares — split (per-column buffers, the
+  // current kernel) or a legacy packed blob from an older saved state.
   _applyAppend(msg, buffers) {
     const spec = msg.spec;
-    const blobRaw = buffers && buffers[0];
-    if (!spec || !blobRaw || !spec.traces) return;
-    const blob = bytesToSpan(blobRaw);
+    if (!spec || !spec.traces) return;
+    const raw = spec.buffer_layout === "split" ? buffers : buffers && buffers[0];
+    if (raw == null) return;
+    const payload = payloadBuffers(spec, raw);
     // Follow policy, decided against the OLD home view before it moves:
     // - at home (never zoomed, or axes reset): the chart follows its data —
     //   refit both axes to the new domain, the live-dashboard default.
@@ -228,7 +231,7 @@ Object.assign(ChartView.prototype, {
       nextView = { ...this.view, x1: nextHome.x1, x0: nextHome.x1 - w };
     }
     const animated = !!spec.animation || spec.traces.some((trace) => !!trace.animation);
-    if (animated && !this._glLost && this.gl && this.updatePayload(spec, blob)) {
+    if (animated && !this._glLost && this.gl && this.updatePayload(spec, payload)) {
       // updatePayload owns previous/next GPU lifetime and matching. Preserve
       // append's follow policy instead of always animating to the new home
       // domain (history inspection must remain stationary).
@@ -241,7 +244,7 @@ Object.assign(ChartView.prototype, {
     // rebuilds the streamed state, not the initial one.
     this.spec = spec;
     this.axes = this._normalizeAxes(spec);
-    this._payload = blob;
+    this._payload = payload;
     this.view0 = this._copyView({
       ranges: Object.fromEntries(Object.entries(this.axes).map(([id, axis]: any) => [id, [...axis.range]])),
     });
@@ -262,7 +265,7 @@ Object.assign(ChartView.prototype, {
       const ts = spec.traces.find((t) => t.id === id);
       if (i < 0 || !ts) continue;
       this._destroyTraceResources(this.gpuTraces[i], texSeen);
-      this.gpuTraces[i] = this._buildTrace(blob, ts);
+      this.gpuTraces[i] = this._buildTrace(payload, ts);
     }
     this._updatePickable();
     this._scheduleViewRequest(this.view, { delay: 0 });

js/src/60_entries.ts

@@ -1,4 +1,4 @@
-import { bytesToSpan, decodeFrame } from "./00_header";
+import { bytesToSpan, decodeFrame, payloadBuffers, payloadCoherent } from "./00_header";
 import { ChartView } from "./50_chartview";
 import { MARK_KINDS, markOf } from "./55_marks";
 // Prototype-augmentation modules: imported for their side effect of attaching
@@ -15,37 +15,51 @@ import "./57_viewstate";
 // Entry points
 // ---------------------------------------------------------------------------
 
-/** First-paint buffers in the shape the spec declares (§29): packed is one
- * blob; split is one span per column. Aligned views stay zero-copy; only a
- * legacy unaligned view pays a narrow view-sized copy. A spec/transport
- * disagreement is a bug, never a fallback. */
-function payloadBuffers(spec, raw) {
-  if (spec.buffer_layout === "split") {
-    if (!Array.isArray(raw)) {
-      throw new Error("xy: spec says buffer_layout=split but the transport delivered one buffer");
-    }
-    return raw.map(bytesToSpan);
-  }
-  if (Array.isArray(raw)) {
-    throw new Error("xy: transport delivered a buffer list but the spec is not split-layout");
-  }
-  return bytesToSpan(raw);
-}
-
 export function render({ model, el }) {
   const spec = model.get("spec");
   const buffer = payloadBuffers(spec, model.get("buffers"));
   const comm = {
     send: (msg) => model.send(msg),
-    wantsViewChange: () => spec.interaction?._transport_view_change === true,
+    // Read the live spec: appends re-sync it, and the transport flag must
+    // survive them (it is re-applied kernel-side on every append).
+    wantsViewChange: () => model.get("spec")?.interaction?._transport_view_change === true,
     onMessage: (cb) => {
       const handler = (content, buffers) => cb(content, buffers);
       model.on("msg:custom", handler);
       return () => model.off?.("msg:custom", handler);
     },
   };
   const view = new ChartView(el, spec, buffer, comm);
-  return () => view.destroy();
+  // Streaming append rides the spec+buffers trait update itself (§29): one
+  // comm message per tick that doubles as notebook-reopen state. A fresh
+  // render above already painted the streamed state, so only a *subsequent*
+  // advance of `spec.append.seq` applies as an incremental append.
+  //
+  // Hosts are not guaranteed to set the two traits atomically: one change
+  // event may fire between the spec write and the buffers write (in either
+  // order). Both events funnel here, a torn pair (a column that no longer
+  // fits its buffer) defers without consuming the seq, and applied state is
+  // keyed on (seq, buffers identity) — so if a same-length torn pair slips
+  // past the fit check, the buffers' own change event re-applies and repairs.
+  const applied = { seq: spec.append?.seq ?? null, buffers: model.get("buffers") };
+  const onAppendState = () => {
+    const nextSpec = model.get("spec");
+    const tag = nextSpec?.append;
+    if (!tag) return;
+    const nextBuffers = model.get("buffers");
+    if (tag.seq === applied.seq && nextBuffers === applied.buffers) return;
+    if (!payloadCoherent(nextSpec, nextBuffers)) return; // torn: wait for the pair
+    applied.seq = tag.seq;
+    applied.buffers = nextBuffers;
+    view._applyAppend({ type: "append", affected: tag.affected, spec: nextSpec }, nextBuffers);
+  };
+  model.on("change:spec", onAppendState);
+  model.on("change:buffers", onAppendState);
+  return () => {
+    model.off?.("change:spec", onAppendState);
+    model.off?.("change:buffers", onAppendState);
+    view.destroy();
+  };
 }
 
 /** Standalone (static HTML export — no kernel). Retains typed CPU views of

python/xy/_figure.py

@@ -157,6 +157,10 @@ def __init__(
         # `view_state()` never round-trips. Reads are eventually consistent.
         self._view_state_ranges: Optional[dict[str, list[float]]] = None
         self._view_state_selection: Optional[dict[str, Any]] = None
+        # Monotonic streaming-append counter; rides the spec as
+        # `append.seq` so trait-transported hosts can detect the refresh
+        # (wire-protocol §4).
+        self._append_seq = 0
 
     # -- axis config --------------------------------------------------------
 

python/xy/config.py

@@ -8,7 +8,9 @@
 from __future__ import annotations
 
 # Wire protocol version: the client refuses a mismatched spec loudly (§33).
-PROTOCOL_VERSION = 4
+# v5: streaming append ships split-layout buffers and, on the widget host,
+# rides the spec/buffers trait update (`spec.append.seq`) with no custom send.
+PROTOCOL_VERSION = 5
 
 # Line traces longer than this ship M4-decimated (Tier 1, §5); the canonical
 # column stays kernel-side for re-decimation on zoom (§28: recompute for the

python/xy/interaction.py

@@ -675,16 +675,20 @@ def append_data(
     alpha: Any = None,
     stroke_width: Any = None,
     symbol: Any = None,
-) -> tuple[dict[str, Any], list[bytes]]:
+) -> tuple[dict[str, Any], list[memoryview]]:
     """Streaming append (Phase-0): extend a trace's canonical
     columns in place and return the client refresh message.
 
     The wire never ships deltas because it never needs to: every tier's payload
     is screen-bounded by construction (design dossier §29 — direct ≤ budget,
     M4 ≤ 4·px, density = grid), so re-emitting the affected trace costs
-    O(pixels), not O(N). The message carries a complete fresh payload; the
-    client rebuilds only the traces named in `affected` and re-requests its
-    current view through the normal stale-while-revalidate path.
+    O(pixels), not O(N). The message carries a complete fresh payload in the
+    split layout (per-column borrowed views, no join copy); the client rebuilds
+    only the traces named in `affected` and re-requests its current view
+    through the normal stale-while-revalidate path. The spec carries
+    `append: {seq, affected}` so a host whose transport is the payload itself
+    (the widget's spec+buffers trait update) can detect and apply the refresh
+    without a separate message envelope.
 
     Phase-0 contract (violations raise before anything mutates):
     - scatter and line traces only;
@@ -863,5 +867,13 @@ def _style_tail(name: str, values: Any) -> Optional[np.ndarray]:
         t._pyr_handle = None
     lod.exit_drill(t)
 
-    spec, blob = fig.build_payload()
-    return {"type": "append", "affected": [t.id], "spec": spec}, [blob]
+    # Split layout, same as first paint (§29): per-column borrowed views, no
+    # join copy. The spec itself names the append — `append.seq` is the apply
+    # signal for the widget host, where the refresh rides the spec+buffers
+    # trait update as one comm message that doubles as notebook-reopen state.
+    # The socket.io host wraps the same spec in an `append` message push.
+    fig._append_seq += 1
+    seq = fig._append_seq
+    spec, buffers = fig.build_payload_split()
+    spec["append"] = {"seq": seq, "affected": [t.id]}
+    return {"type": "append", "affected": [t.id], "spec": spec}, buffers