perf(widgets): render async data incrementally and match main update rates

bec_widgets/widgets/plots/image/image.py

@@ -130,6 +130,7 @@ class Image(ImageBase):
         self._data_bridge: QtDataSubscription | None = None
         self._source_key: tuple[str, str] | None = None
         self._min_display_ordinal: int | None = None
+        self._waterfall_cache: dict | None = None
         self.old_scan_id = None
         self.scan_id = None
         self.async_update = False
@@ -451,7 +452,7 @@ class Image(ImageBase):
                 sources=[(self._config.device, entry)],
                 scan=scan,
                 parent=self,
-                min_emit_interval=0.1,
+                min_emit_interval=0.04,
                 max_points=max_points,
             )
             self._data_bridge.updated.connect(self._on_data_update)
@@ -466,6 +467,7 @@ class Image(ImageBase):
 
         self._source_key = (self._config.device, entry)
         self._min_display_ordinal = None
+        self._waterfall_cache = None
         self._set_connection_status("connected")
         logger.info(
             f"Connected to {self._config.device}.{self._config.signal} with type {config.monitor_type}"
@@ -476,6 +478,7 @@ class Image(ImageBase):
         """Close the active DataAPI bridge, if any."""
         self._source_key = None
         self._min_display_ordinal = None
+        self._waterfall_cache = None
         if self._data_bridge is None:
             return
         try:
@@ -752,7 +755,7 @@ class Image(ImageBase):
         if self.subscriptions["main"].monitor_type == "2d":
             data = np.asarray(source.values[-1])
         else:
-            data = self._build_1d_buffer(source)
+            data = self._build_1d_buffer(source, reason=update.reason)
         if data is None:
             return
         self._render_image_data(data)
@@ -801,14 +804,85 @@ class Image(ImageBase):
         if self.crosshair is not None:
             self.crosshair.reset()
 
-    def _build_1d_buffer(self, source) -> np.ndarray | None:
+    def _build_1d_buffer(self, source, reason: str = "live") -> np.ndarray | None:
         """
-        Rebuild the 2-D waterfall buffer from the 1-D columnar fragments of a
+        Build the 2-D waterfall buffer from the 1-D columnar fragments of a
         source: one row per ordinal, rows zero-padded to the longest row,
         newest row last. Covers async 'add' (one fragment per ordinal),
         'add_slice' (accumulated row per ordinal), 'replace' (single current
         state) and preview streams (one waveform per arrival) alike.
 
+        The padded buffer and the consumed ordinal frontier are cached, so a
+        live append-only emission stacks only the new rows (padded to the
+        cached width) — O(new data) per emission instead of O(total). The
+        buffer is rebuilt from all fragments when the emission cannot be a
+        pure append: a non-live reason, a scan or display-window change, new
+        data at or below the frontier (late hole-fills, retention drops) or a
+        new row wider than the cached buffer. Every full rebuild reseeds the
+        cache.
+
+        Args:
+            source (SourceData): The 1-D source snapshot.
+            reason (str): The update reason ("live", "backfill", ...).
+
+        Returns:
+            np.ndarray | None: The (n_rows, max_len) buffer, or None if no
+                displayable rows remain.
+        """
+        ordinals = source.ordinals
+        cache = self._waterfall_cache
+        if (
+            cache is not None
+            and reason == "live"
+            and ordinals
+            and cache["scan_id"] == self.scan_id
+            and cache["min_display_ordinal"] == self._min_display_ordinal
+        ):
+            n_seen = cache["n_seen"]
+            frontier_intact = (
+                len(ordinals) >= n_seen and ordinals[n_seen - 1] == cache["last_ordinal"]
+            )
+            if frontier_intact and len(ordinals) == n_seen:
+                # Unchanged snapshot (the backend reuses source snapshots).
+                return cache["buffer"]
+            if frontier_intact and ordinals[n_seen] > cache["last_ordinal"]:
+                new_rows = [np.atleast_1d(np.asarray(value)) for value in source.values[n_seen:]]
+                new_rows = [row for row in new_rows if row.ndim == 1]
+                width = cache["width"]
+                if all(row.shape[0] <= width for row in new_rows):
+                    buffer = cache["buffer"]
+                    if new_rows:
+                        padded = [
+                            np.pad(
+                                row, (0, width - row.shape[0]), mode="constant", constant_values=0
+                            )
+                            for row in new_rows
+                        ]
+                        buffer = np.vstack([buffer, *padded])
+                        cache["buffer"] = buffer
+                    cache["n_seen"] = len(ordinals)
+                    cache["last_ordinal"] = ordinals[-1]
+                    return buffer
+        buffer = self._rebuild_1d_buffer(source)
+        if ordinals:
+            self._waterfall_cache = {
+                "scan_id": self.scan_id,
+                "min_display_ordinal": self._min_display_ordinal,
+                "n_seen": len(ordinals),
+                "last_ordinal": ordinals[-1],
+                "buffer": buffer,
+                "width": 0 if buffer is None else buffer.shape[1],
+            }
+        else:
+            self._waterfall_cache = None
+        return buffer
+
+    def _rebuild_1d_buffer(self, source) -> np.ndarray | None:
+        """
+        From-scratch reference construction of the waterfall buffer (see
+        :meth:`_build_1d_buffer`): all fragments, zero-padded to the longest
+        row, restricted to the current display window.
+
         Args:
             source (SourceData): The 1-D source snapshot.
 

bec_widgets/widgets/plots/motor_map/motor_map.py

@@ -672,7 +672,7 @@ class MotorMap(PlotBase):
                 sources=[(device_x, device_x), (device_y, device_y)],
                 scan=None,
                 parent=self,
-                min_emit_interval=0.1,
+                min_emit_interval=0.04,
                 max_points=self.config.max_points,
             )
             self._data_bridge.updated.connect(self._on_data_update)

bec_widgets/widgets/plots/multi_waveform/multi_waveform.py

@@ -616,7 +616,7 @@ class MultiWaveform(PlotBase):
         if self._source_key is None:
             return
         source = update.get(*self._source_key)
-        if source is None or not source.values:
+        if source is None or source.values is None or len(source.values) == 0:
             return
 
         current_scan_id = self._effective_scan_id(update, source)
@@ -835,7 +835,7 @@ class MultiWaveform(PlotBase):
                 sources=[(device, entry)],
                 scan=scan,
                 parent=self,
-                min_emit_interval=0.1,
+                min_emit_interval=0.04,
                 max_points=max_points,
             )
             self._data_bridge.updated.connect(self._on_data_update)

bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py

@@ -345,7 +345,7 @@ class ScatterWaveform(PlotBase):
             return
         try:
             self._data_bridge = QtDataSubscription(
-                self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.1
+                self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.04
             )
             self._data_bridge.updated.connect(self._on_data_update)
         except Exception as exc:

bec_widgets/widgets/plots/waveform/waveform.py

@@ -1748,8 +1748,10 @@ class Waveform(PlotBase):
         if not sources:
             return
         try:
+            # 15 Hz render coalescing: above typical device message rates while
+            # leaving paint headroom for multi-million-point curves.
             self._data_bridge = QtDataSubscription(
-                self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.1
+                self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.0667
             )
             self._data_bridge.updated.connect(self._on_data_update)
         except Exception as exc:
@@ -1793,7 +1795,7 @@ class Waveform(PlotBase):
                     sources=list(dict.fromkeys(sources)),
                     scan=scan_id,
                     parent=self,
-                    min_emit_interval=0.1,
+                    min_emit_interval=0.0667,
                 )
                 bridge.updated.connect(self._on_data_update)
                 self._history_bridges[scan_id] = bridge
@@ -2040,7 +2042,7 @@ class Waveform(PlotBase):
         lengths only) or a same-length x device column, with an index
         fallback on any length mismatch.
         """
-        y_data = self._async_display_values(source)
+        y_data = self._async_display_values_cached(curve, update, source)
         if y_data is None or len(y_data) == 0:
             return False
         mode = self.x_axis_mode["name"] or "auto"
@@ -2093,7 +2095,7 @@ class Waveform(PlotBase):
             np.ndarray | None: The displayed y data.
         """
         values = source.values
-        if not values:
+        if values is None or len(values) == 0:
             return None
         update_type = source.metadata.get("async_update_type")
         max_shape = source.metadata.get("max_shape") or []
@@ -2113,6 +2115,75 @@ class Waveform(PlotBase):
             return np.asarray(values)
         return np.atleast_1d(np.asarray(values[-1]))
 
+    def _async_display_values_cached(self, curve: Curve, update, source) -> np.ndarray | None:
+        """
+        Incremental variant of :meth:`_async_display_values` for the one
+        display mode whose from-scratch cost grows with the scan — 1-D 'add'
+        concatenation. The concatenated buffer and the consumed ordinal
+        frontier are cached on the curve; per emission only the fragments
+        beyond the frontier are appended, keeping the per-message cost
+        O(new data) instead of O(total).
+
+        The cache is dropped and the series rebuilt from all fragments when
+        the emission cannot be a pure append: a non-live reason (backfill,
+        history, rebind), a scan change, a source-key change, or new data at
+        or below the frontier (late hole-fills, retention drops). Every full
+        rebuild reseeds the cache, so a live stream resumes incrementally
+        after it. All other display modes are already O(new data) and are
+        delegated unchanged.
+
+        Args:
+            curve(Curve): The rendered curve (cache carrier).
+            update(SubscriptionUpdate): The update snapshot.
+            source(SourceData): The async source snapshot.
+
+        Returns:
+            np.ndarray | None: The displayed y data (identical to
+                :meth:`_async_display_values`).
+        """
+        values = source.values
+        if values is None or len(values) == 0:
+            return None
+        update_type = source.metadata.get("async_update_type")
+        max_shape = source.metadata.get("max_shape") or []
+        if update_type != "add" or len(max_shape) > 1:
+            # Last-fragment / last-row display modes: O(new data) already.
+            curve._data_api_async_cache = None
+            return self._async_display_values(source)
+        ordinals = source.ordinals
+        if ordinals is None or len(ordinals) == 0:
+            return self._async_display_values(source)
+        cache = getattr(curve, "_data_api_async_cache", None)
+        if (
+            cache is not None
+            and update.reason == "live"
+            and cache["scan_id"] == update.scan_id
+            and cache["source_key"] == source.key
+        ):
+            n_seen = cache["n_seen"]
+            frontier_intact = (
+                len(ordinals) >= n_seen and ordinals[n_seen - 1] == cache["last_ordinal"]
+            )
+            if frontier_intact and len(ordinals) == n_seen:
+                # Unchanged snapshot (the backend reuses source snapshots).
+                return cache["buffer"]
+            if frontier_intact and ordinals[n_seen] > cache["last_ordinal"]:
+                new_fragments = [np.atleast_1d(np.asarray(value)) for value in values[n_seen:]]
+                buffer = np.concatenate([cache["buffer"], *new_fragments])
+                cache["buffer"] = buffer
+                cache["n_seen"] = len(ordinals)
+                cache["last_ordinal"] = ordinals[-1]
+                return buffer
+        buffer = self._async_display_values(source)
+        curve._data_api_async_cache = {
+            "scan_id": update.scan_id,
+            "source_key": source.key,
+            "n_seen": len(ordinals),
+            "last_ordinal": ordinals[-1],
+            "buffer": buffer,
+        }
+        return buffer
+
     def _auto_adjust_async_curve_settings(
         self,
         curve: Curve,

tests/unit_tests/benchmarks/test_data_api_widget_throughput.py

@@ -0,0 +1,161 @@
+"""
+Comparative throughput benchmark of the DataAPI widget rendering path.
+
+Feeds one async 'add' source through the waveform's ``_on_data_update``
+(``curve.setData`` mocked out, isolating the data-path cost) and compares it
+with a main-style baseline: one ``np.hstack`` of (buffer, fragment) per
+readback message for the same total data. The DataAPI path receives the data
+in coalesced emissions (10 fragments per emission) and renders each one
+incrementally, so it must not be slower than the per-message baseline.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from unittest.mock import MagicMock
+
+import numpy as np
+import pytest
+from bec_lib.data_api.models import SourceData, SubscriptionUpdate
+
+from bec_widgets.widgets.plots.waveform.waveform import Waveform
+from tests.unit_tests.client_mocks import mocked_client
+from tests.unit_tests.conftest import create_widget
+
+N_EMISSIONS = 150
+FRAGMENTS_PER_EMISSION = 10
+SAMPLES_PER_FRAGMENT = 500
+#: Timing rounds per path; the best round is compared (filters scheduler noise).
+ROUNDS = 2
+
+
+@pytest.fixture
+def waveform_widget(qtbot, mocked_client, monkeypatch):
+    """A Waveform with a stubbed data bridge and one async curve."""
+
+    def factory(client, sources, scan="live", parent=None, **kwargs):
+        bridge = MagicMock()
+        bridge.sources = list(sources)
+        bridge.scan_id = None if scan == "live" else scan
+        bridge.healthy = True
+        return bridge
+
+    monkeypatch.setattr("bec_widgets.widgets.plots.waveform.waveform.QtDataSubscription", factory)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    yield wf
+
+
+def _build_updates(fragments: list[np.ndarray]) -> list[SubscriptionUpdate]:
+    """
+    One SubscriptionUpdate per emission, each snapshot extending the previous
+    one by FRAGMENTS_PER_EMISSION fragments. The fragment objects are shared
+    between snapshots, mimicking the backend's snapshot reuse.
+    """
+    updates = []
+    metadata = {"async_update_type": "add", "max_shape": [None], "acquisition_group": None}
+    for emission in range(N_EMISSIONS):
+        n_fragments = (emission + 1) * FRAGMENTS_PER_EMISSION
+        ordinals = tuple(range(n_fragments))
+        source = SourceData(
+            device="async_device",
+            entry="async_device",
+            kind="async",
+            ordinals=ordinals,
+            values=tuple(fragments[:n_fragments]),
+            timestamps=tuple(float(i) for i in ordinals),
+            complete=True,
+            metadata=metadata,
+        )
+        updates.append(
+            SubscriptionUpdate(
+                scan_id="benchmark_scan",
+                reason="live",
+                sources={source.key: source},
+                aligned_ordinals=ordinals,
+                complete=True,
+                metadata={"group": "scan"},
+            )
+        )
+    return updates
+
+
+def _measure_dataapi(wf: Waveform, curve, updates: list[SubscriptionUpdate]) -> float:
+    """Time one full pass of the DataAPI path over all emissions."""
+    curve._data_api_async_cache = None  # each pass starts from an empty cache
+    start = time.perf_counter()
+    for update in updates:
+        wf._on_data_update(update)
+    return time.perf_counter() - start
+
+
+def _measure_baseline(fragments: list[np.ndarray]) -> tuple[float, np.ndarray]:
+    """Time main's per-message loop: one np.hstack of (buffer, fragment) each."""
+    start = time.perf_counter()
+    buffer = np.empty(0)
+    for fragment in fragments:
+        buffer = np.hstack((buffer, fragment))
+    return time.perf_counter() - start, buffer
+
+
+def test_data_api_waveform_throughput(waveform_widget, monkeypatch):
+    """
+    The DataAPI rendering path (coalesced emissions, incremental append) must
+    match the per-message cost of main's incremental hstack loop for the same
+    total data.
+    """
+    wf = waveform_widget
+    curve = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf.scan_id = "benchmark_scan"
+    wf.x_axis_mode["name"] = "index"
+    monkeypatch.setattr(curve, "setData", lambda *args, **kwargs: None)
+
+    rng = np.random.default_rng(42)
+    total_fragments = N_EMISSIONS * FRAGMENTS_PER_EMISSION
+    fragments = [rng.random(SAMPLES_PER_FRAGMENT) for _ in range(total_fragments)]
+    updates = _build_updates(fragments)
+
+    # The mocked BEC client keeps polling threads alive; every numpy GIL
+    # release then stalls the timed loop for up to the thread switch interval
+    # (5 ms by default), swamping the actual computation. A short interval
+    # during the measurement removes that scheduler noise for both paths.
+    switch_interval = sys.getswitchinterval()
+    sys.setswitchinterval(1e-4)
+    try:
+        # Warm-up: fault in pages and settle CPU scheduling before timing.
+        _measure_baseline(fragments[: total_fragments // 5])
+        _measure_dataapi(wf, curve, updates[: N_EMISSIONS // 5])
+
+        # Alternate the measurement order between rounds and keep the best
+        # round of each path.
+        dataapi_times: list[float] = []
+        baseline_times: list[float] = []
+        baseline_buffer = None
+        for round_index in range(ROUNDS):
+            if round_index % 2 == 0:
+                baseline_time, baseline_buffer = _measure_baseline(fragments)
+                baseline_times.append(baseline_time)
+                dataapi_times.append(_measure_dataapi(wf, curve, updates))
+            else:
+                dataapi_times.append(_measure_dataapi(wf, curve, updates))
+                baseline_time, baseline_buffer = _measure_baseline(fragments)
+                baseline_times.append(baseline_time)
+    finally:
+        sys.setswitchinterval(switch_interval)
+    dataapi_time = min(dataapi_times)
+    baseline_time = min(baseline_times)
+
+    # Both paths must have produced the identical series.
+    rendered = wf._async_display_values_cached(
+        curve, updates[-1], updates[-1].sources[("async_device", "async_device")]
+    )
+    np.testing.assert_array_equal(rendered, baseline_buffer)
+
+    print(
+        f"\nDataAPI path: {dataapi_time * 1000:.1f} ms for {N_EMISSIONS} emissions "
+        f"({total_fragments} fragments, {baseline_buffer.size} samples); "
+        f"main-style per-message baseline: {baseline_time * 1000:.1f} ms"
+    )
+    assert dataapi_time <= baseline_time * 1.5, (
+        f"DataAPI rendering path too slow: {dataapi_time:.3f}s vs " f"baseline {baseline_time:.3f}s"
+    )

tests/unit_tests/test_image_view_next_gen.py

@@ -741,6 +741,69 @@ def test_scan_rollover_same_scan_noop(qtbot, mocked_client, monkeypatch):
     assert view.main_image.raw_data.shape == (2, 3)
 
 
+def test_build_1d_buffer_incremental_matches_full_rebuild(qtbot, mocked_client, monkeypatch):
+    """
+    The incremental waterfall buffer must equal the from-scratch construction
+    across live appends (contiguous and gapped), an unchanged reused
+    snapshot, an out-of-order hole-fill, a wider new row, a non-live reason
+    and a scan change — and must only fall back to the full rebuild for the
+    emissions that cannot be pure appends.
+    """
+    _fake_bridge_factory(monkeypatch)
+    view = create_widget(qtbot, Image, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj"
+    )
+    view.image(device="eiger", signal="img")
+
+    from_scratch = Image._rebuild_1d_buffer
+    rebuilds = []
+
+    def counting(self, source):
+        rebuilds.append(source)
+        return from_scratch(self, source)
+
+    monkeypatch.setattr(Image, "_rebuild_1d_buffer", counting)
+
+    def reference(values):
+        rows = [np.atleast_1d(np.asarray(value)) for value in values]
+        width = max(row.shape[0] for row in rows)
+        return np.vstack([np.pad(row, (0, width - row.shape[0])) for row in rows])
+
+    steps = [
+        # (scan_id, reason, values, ordinals, expected rebuild count so far)
+        ("scan_1", "live", [[1.0, 2.0]], (0,), 1),  # first emission seeds the cache
+        ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1),  # append
+        ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1),  # unchanged reused snapshot
+        ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [5.0]], (0, 1, 3), 1),  # gapped short append
+        # late hole-fill below the frontier -> full rebuild
+        ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0]], (0, 1, 2, 3), 2),
+        # new row wider than the cached buffer -> full rebuild
+        ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4], (0, 1, 2, 3, 4), 3),
+        # non-live reason -> full rebuild
+        (
+            "scan_1",
+            "backfill",
+            [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4],
+            (0, 1, 2, 3, 4),
+            4,
+        ),
+        ("scan_2", "live", [[9.0, 9.0]], (0,), 5),  # scan change -> full rebuild
+        ("scan_2", "live", [[9.0, 9.0], [10.0, 11.0]], (0, 1), 5),  # incremental resumes
+    ]
+    for scan_id, reason, values, ordinals, expected_rebuilds in steps:
+        source = _make_source(
+            "eiger",
+            "async_obj",
+            values,
+            ordinals=ordinals,
+            metadata={"async_update_type": "add", "max_shape": [None]},
+        )
+        view._on_data_update(_make_update(source, scan_id=scan_id, reason=reason))
+        np.testing.assert_array_equal(view.main_image.raw_data, reference(values))
+        assert len(rebuilds) == expected_rebuilds
+
+
 def test_image_data_update_2d(qtbot, mocked_client, monkeypatch):
     _fake_bridge_factory(monkeypatch)
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)

tests/unit_tests/test_waveform.py

@@ -113,20 +113,21 @@ def _fake_bridge_factory(monkeypatch, gated_bytes: int | None = None):
     return created
 
 
-def _monitored_source(device, values, entry=None, timestamps=None, ordinals=None):
+def _monitored_source(device, values, entry=None, timestamps=None, ordinals=None, as_numpy=False):
     from bec_lib.data_api.models import SourceData
 
     entry = entry or device
     ordinals = tuple(range(len(values))) if ordinals is None else tuple(ordinals)
     if timestamps is None:
         timestamps = tuple(float(i) for i in ordinals)
+    wrap = (lambda seq: np.asarray(seq)) if as_numpy else tuple
     return SourceData(
         device=device,
         entry=entry,
         kind="monitored",
-        ordinals=ordinals,
-        values=tuple(values),
-        timestamps=tuple(timestamps),
+        ordinals=wrap(ordinals),
+        values=wrap(values),
+        timestamps=wrap(timestamps),
         complete=True,
     )
 
@@ -140,6 +141,7 @@ def _async_source(
     max_shape=(None,),
     kind="async",
     ordinals=None,
+    as_numpy=False,
 ):
     from bec_lib.data_api.models import SourceData
 
@@ -147,13 +149,14 @@ def _async_source(
     ordinals = tuple(range(len(values))) if ordinals is None else tuple(ordinals)
     if timestamps is None:
         timestamps = tuple(float(i) for i in ordinals)
+    wrap = (lambda seq: np.asarray(seq)) if as_numpy else tuple
     return SourceData(
         device=device,
         entry=entry,
         kind=kind,
-        ordinals=ordinals,
-        values=tuple(values),
-        timestamps=tuple(timestamps),
+        ordinals=wrap(ordinals),
+        values=wrap(values),
+        timestamps=wrap(timestamps),
         complete=True,
         metadata={
             "async_update_type": update_type,
@@ -1324,6 +1327,55 @@ def test_on_data_update_async_history_rows(qtbot, mocked_client, monkeypatch):
     np.testing.assert_array_equal(y_data, [4, 5, 6])
 
 
+def test_on_data_update_async_add_incremental_matches_full_rebuild(
+    qtbot, mocked_client, monkeypatch
+):
+    """
+    The incremental 1-D 'add' render path must yield exactly the series the
+    from-scratch concatenation yields, across live appends (contiguous and
+    gapped), an unchanged reused snapshot, an out-of-order hole-fill, a
+    non-live reason and a scan change — and must only fall back to the full
+    rebuild for the emissions that cannot be pure appends.
+    """
+    _fake_bridge_factory(monkeypatch)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    c = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf.scan_id = None  # render updates of any scan
+    wf.x_axis_mode["name"] = "index"
+
+    from_scratch = Waveform._async_display_values
+    rebuilds = []
+
+    def counting(source):
+        rebuilds.append(source)
+        return from_scratch(source)
+
+    monkeypatch.setattr(Waveform, "_async_display_values", staticmethod(counting))
+
+    steps = [
+        # (scan_id, reason, values, ordinals, expected rebuild count so far)
+        ("scan_1", "live", ([0.0, 1.0],), (0,), 1),  # first emission seeds the cache
+        ("scan_1", "live", ([0.0, 1.0], [2.0]), (0, 1), 1),  # append
+        ("scan_1", "live", ([0.0, 1.0], [2.0]), (0, 1), 1),  # unchanged reused snapshot
+        ("scan_1", "live", ([0.0, 1.0], [2.0], [4.0, 5.0]), (0, 1, 3), 1),  # gapped append
+        # late hole-fill below the frontier -> full rebuild
+        ("scan_1", "live", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0]), (0, 1, 2, 3), 2),
+        ("scan_1", "live", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0], [6.0]), (0, 1, 2, 3, 4), 2),
+        # non-live reason -> full rebuild
+        ("scan_1", "backfill", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0], [6.0]), tuple(range(5)), 3),
+        ("scan_2", "live", ([7.0],), (0,), 4),  # scan change -> full rebuild
+        ("scan_2", "live", ([7.0], [8.0, 9.0]), (0, 1), 4),  # incremental resumes
+    ]
+    for scan_id, reason, values, ordinals, expected_rebuilds in steps:
+        source = _async_source("async_device", values=values, ordinals=ordinals, update_type="add")
+        wf._on_data_update(_make_update([source], scan_id=scan_id, reason=reason))
+        x_data, y_data = c.get_data()
+        expected = from_scratch(source)
+        np.testing.assert_array_equal(y_data, expected)
+        np.testing.assert_array_equal(x_data, np.arange(len(expected)))
+        assert len(rebuilds) == expected_rebuilds
+
+
 ##################################################
 # The following tests are for the Curve class
 ##################################################
@@ -2125,3 +2177,44 @@ def test_detector_shaped_history_curve_is_not_hidden(qtbot, mocked_client, monke
     scan_item._msg.num_monitored_readouts = 37
     scan_item._msg.num_points = 37
     assert wf._history_curve_compatible(curve_for("bpm4i", "bpm4i")) is False
+
+
+def test_history_source_with_numpy_columns_renders(qtbot, mocked_client, monkeypatch):
+    """Regression: the history plugin delivers numpy-array columns (bulk
+    ingest keeps the file arrays intact). The render must accept them; a
+    ``if not source.values`` truth-test raised ValueError on multi-element
+    arrays and SafeSlot swallowed it, so the curve showed no data."""
+    _fake_bridge_factory(monkeypatch)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    wf.x_axis_mode["name"] = "index"
+
+    # monitored history column, numpy-valued (what bec_lib now emits)
+    c = wf.plot(arg1="bpm4i", label="bpm4i-bpm4i")
+    wf.scan_id = "dummy"
+    src = _monitored_source("bpm4i", values=[5.0, 6.0, 7.0, 8.0], as_numpy=True)
+    wf._on_data_update(_make_update([src], reason="history"))
+    x_data, y_data = c.get_data()
+    np.testing.assert_array_equal(y_data, [5.0, 6.0, 7.0, 8.0])
+
+    # async history waveform, flat numpy value column, no async_update_type
+    ca = wf.plot(arg1="async_device", label="async_device-async_device")
+    src_a = _async_source(
+        "async_device", values=[1.0, 2.0, 3.0, 4.0, 5.0], update_type=None, as_numpy=True
+    )
+    # history reads carry no async_update_type
+    src_a.metadata.pop("async_update_type", None)
+    wf._on_data_update(_make_update([src_a], reason="history"))
+    x_data, y_data = ca.get_data()
+    assert y_data is not None and len(y_data) == 5
+
+
+def test_async_display_values_accepts_numpy(qtbot, mocked_client):
+    """Both display helpers must handle numpy-valued sources (bulk history)."""
+    from types import SimpleNamespace
+
+    src = _async_source("async_device", values=[1, 2, 3], update_type="add", as_numpy=True)
+    assert list(Waveform._async_display_values(src)) == [1, 2, 3]
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    carrier = SimpleNamespace(_data_api_async_cache=None)
+    cached = wf._async_display_values_cached(carrier, _make_update([src]), src)
+    assert list(cached) == [1, 2, 3]