diff --git a/bec_widgets/widgets/plots/image/image.py b/bec_widgets/widgets/plots/image/image.py
index 0b5dd483..a9e2f077 100644
--- a/bec_widgets/widgets/plots/image/image.py
+++ b/bec_widgets/widgets/plots/image/image.py
@@ -5,7 +5,6 @@ from typing import Literal
 
 import numpy as np
 from bec_lib import bec_logger
-from bec_lib.endpoints import MessageEndpoints
 from pydantic import BaseModel, Field, field_validator
 from qtpy.QtCore import QTimer
 from qtpy.QtWidgets import QWidget
@@ -13,6 +12,7 @@ from qtpy.QtWidgets import QWidget
 from bec_widgets.utils.bec_connector import ConnectionConfig
 from bec_widgets.utils.colors import Colors, apply_theme
 from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
+from bec_widgets.utils.qt_data_subscription import QtDataSubscription
 from bec_widgets.widgets.plots.image.image_base import ImageBase
 from bec_widgets.widgets.plots.image.image_item import ImageItem
 from bec_widgets.widgets.plots.image.toolbar_components.device_selection import (
@@ -106,6 +106,11 @@ class Image(ImageBase):
 
     SUPPORTED_SIGNALS = ["AsyncSignal", "AsyncMultiSignal", "DynamicSignal"]
 
+    #: Retention cap for scan-less 1D preview streams (rows of the waterfall buffer).
+    PREVIEW_1D_MAX_ROWS = 1000
+    #: Retention cap for scan-less 2D preview streams (only the newest frame is shown).
+    PREVIEW_2D_MAX_FRAMES = 2
+
     def __init__(
         self,
         parent: QWidget | None = None,
@@ -121,6 +126,13 @@ class Image(ImageBase):
         self.subscriptions: defaultdict[str, ImageLayerConfig] = defaultdict(ImageLayerConfig)
         # Store signal configs separately (not serialized to QSettings)
         self._signal_configs: dict[str, dict] = {}
+        # Data delivery through the DataAPI (one bridge for the main layer).
+        self._data_bridge: QtDataSubscription | None = None
+        self._source_key: tuple[str, str] | None = None
+        self._min_display_ordinal: int | None = None
+        self.old_scan_id = None
+        self.scan_id = None
+        self.async_update = False
 
         super().__init__(
             parent=parent, config=config, client=client, gui_id=gui_id, popups=popups, **kwargs
@@ -129,11 +141,6 @@ class Image(ImageBase):
         self._autorange_on_next_update = False
         self._init_toolbar_image()
         self.layer_removed.connect(self._on_layer_removed)
-        self.old_scan_id = None
-        self.scan_id = None
-        self.async_update = False
-        self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
-        self.bec_dispatcher.connect_slot(self.on_scan_progress, MessageEndpoints.scan_progress())
 
     @property
     def _config(self) -> ImageLayerConfig:
@@ -271,15 +278,8 @@ class Image(ImageBase):
 
         old_device = self._config.device
         old_signal = self._config.signal
-        old_config = self.subscriptions["main"]
         if old_device and old_signal and old_device != value:
-            self._disconnect_monitor_subscription(
-                device=old_device,
-                signal=old_signal,
-                source=old_config.source,
-                async_update=self.async_update,
-                async_signal_name=old_config.async_signal_name,
-            )
+            self._cleanup_data_api_subscription()
         self._config.device = value
 
         # If we have a signal, reconnect with the new device
@@ -336,15 +336,8 @@ class Image(ImageBase):
             return
 
         old_signal = self._config.signal
-        old_config = self.subscriptions["main"]
         if self._config.device and old_signal and old_signal != value:
-            self._disconnect_monitor_subscription(
-                device=self._config.device,
-                signal=old_signal,
-                source=old_config.source,
-                async_update=self.async_update,
-                async_signal_name=old_config.async_signal_name,
-            )
+            self._cleanup_data_api_subscription()
         self._config.signal = value
 
         # If we have a device, try to connect
@@ -381,7 +374,8 @@ class Image(ImageBase):
 
     def _setup_connection(self):
         """
-        Internal method to setup connection based on current device, signal, and signal_config.
+        Internal method to setup the DataAPI subscription based on current
+        device, signal, and signal_config.
         """
         if not self._config.device or not self._config.signal:
             logger.warning("Cannot setup connection without both device and signal")
@@ -396,8 +390,8 @@ class Image(ImageBase):
             self._set_connection_status("error", "Missing signal config")
             return
 
-        # Disconnect any existing monitor first
-        self._disconnect_current_monitor()
+        # Close any existing subscription first
+        self._cleanup_data_api_subscription()
 
         # Determine monitor type and source from signal_config
         signal_class = signal_config.get("signal_class", None)
@@ -422,123 +416,81 @@ class Image(ImageBase):
             self._set_connection_status("error", "Missing ndim in signal_info")
             return
 
-        config = self.subscriptions["main"]
-        self.async_update = False
-        config.async_signal_name = None
-
-        if ndim == 1:
-            config.source = "device_monitor_1d"
-            config.monitor_type = "1d"
-            if signal_class == "PreviewSignal":
-                self.bec_dispatcher.connect_slot(
-                    self.on_image_update_1d,
-                    MessageEndpoints.device_preview(self._config.device, self._config.signal),
-                )
-            elif signal_class in self.SUPPORTED_SIGNALS:
-                self.async_update = True
-                config.async_signal_name = signal_config.get(
-                    "obj_name", f"{self._config.device}_{self._config.signal}"
-                )
-                self._setup_async_image(self.scan_id)
-        elif ndim == 2:
-            config.source = "device_monitor_2d"
-            config.monitor_type = "2d"
-            if signal_class == "PreviewSignal":
-                self.bec_dispatcher.connect_slot(
-                    self.on_image_update_2d,
-                    MessageEndpoints.device_preview(self._config.device, self._config.signal),
-                )
-            elif signal_class in self.SUPPORTED_SIGNALS:
-                self.async_update = True
-                config.async_signal_name = signal_config.get(
-                    "obj_name", f"{self._config.device}_{self._config.signal}"
-                )
-                self._setup_async_image(self.scan_id)
-        else:
+        if ndim not in (1, 2):
             logger.warning(
                 f"Unsupported ndim '{ndim}' for monitor '{self._config.device}.{self._config.signal}'."
             )
             self._set_connection_status("error", f"Unsupported ndim '{ndim}'")
             return
 
+        config = self.subscriptions["main"]
+        self.async_update = False
+        config.async_signal_name = None
+        config.monitor_type = "1d" if ndim == 1 else "2d"
+        config.source = "device_monitor_1d" if ndim == 1 else "device_monitor_2d"
+
+        if signal_class == "PreviewSignal":
+            # Scan-less device stream served by the DataAPI device plugin.
+            scan = None
+            entry = self._config.signal
+            max_points = self.PREVIEW_1D_MAX_ROWS if ndim == 1 else self.PREVIEW_2D_MAX_FRAMES
+        else:
+            # Scan-scoped async stream; the DataAPI rebinds on new scans and
+            # hands terminal scans over to history automatically.
+            self.async_update = True
+            config.async_signal_name = signal_config.get(
+                "obj_name", f"{self._config.device}_{self._config.signal}"
+            )
+            scan = "live"
+            entry = config.async_signal_name
+            max_points = None
+
+        try:
+            self._data_bridge = QtDataSubscription(
+                self.client,
+                sources=[(self._config.device, entry)],
+                scan=scan,
+                parent=self,
+                min_emit_interval=0.1,
+                max_points=max_points,
+            )
+            self._data_bridge.updated.connect(self._on_data_update)
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(
+                f"Failed to configure image data subscription for "
+                f"{self._config.device}.{self._config.signal}: {exc}"
+            )
+            self._cleanup_data_api_subscription()
+            self._set_connection_status("error", str(exc))
+            return
+
+        self._source_key = (self._config.device, entry)
+        self._min_display_ordinal = None
         self._set_connection_status("connected")
         logger.info(
             f"Connected to {self._config.device}.{self._config.signal} with type {config.monitor_type}"
         )
         self._autorange_on_next_update = True
 
-    def _disconnect_monitor_subscription(
-        self,
-        *,
-        device: str,
-        signal: str,
-        source: Literal["device_monitor_1d", "device_monitor_2d"] | None,
-        async_update: bool,
-        async_signal_name: str | None,
-    ) -> None:
-        if not device or not signal:
+    def _cleanup_data_api_subscription(self):
+        """Close the active DataAPI bridge, if any."""
+        self._source_key = None
+        self._min_display_ordinal = None
+        if self._data_bridge is None:
             return
-
-        if async_update:
-            async_signal_name = async_signal_name or signal
-            ids_to_check = [self.scan_id, self.old_scan_id]
-
-            if source == "device_monitor_1d":
-                for scan_id in ids_to_check:
-                    if scan_id is None:
-                        continue
-                    self.bec_dispatcher.disconnect_slot(
-                        self.on_image_update_1d,
-                        MessageEndpoints.device_async_signal(scan_id, device, async_signal_name),
-                    )
-                    logger.info(
-                        f"Disconnecting 1d update ScanID:{scan_id}, Device Name:{device},Device Entry:{async_signal_name}"
-                    )
-            elif source == "device_monitor_2d":
-                for scan_id in ids_to_check:
-                    if scan_id is None:
-                        continue
-                    self.bec_dispatcher.disconnect_slot(
-                        self.on_image_update_2d,
-                        MessageEndpoints.device_async_signal(scan_id, device, async_signal_name),
-                    )
-                    logger.info(
-                        f"Disconnecting 2d update ScanID:{scan_id}, Device Name:{device},Device Entry:{async_signal_name}"
-                    )
-            return
-
-        if source == "device_monitor_1d":
-            self.bec_dispatcher.disconnect_slot(
-                self.on_image_update_1d, MessageEndpoints.device_preview(device, signal)
-            )
-            logger.info(
-                f"Disconnecting preview 1d update Device Name:{device}, Device Entry:{signal}"
-            )
-        elif source == "device_monitor_2d":
-            self.bec_dispatcher.disconnect_slot(
-                self.on_image_update_2d, MessageEndpoints.device_preview(device, signal)
-            )
-            logger.info(
-                f"Disconnecting preview 2d update Device Name:{device}, Device Entry:{signal}"
-            )
+        try:
+            self._data_bridge.close()
+        finally:
+            self._data_bridge = None
 
     def _disconnect_current_monitor(self):
         """
-        Internal method to disconnect the current monitor subscriptions.
+        Close the current DataAPI subscription and reset the async bookkeeping.
         """
-        if not self._config.device or not self._config.signal:
-            return
-
-        config = self.subscriptions["main"]
-        self._disconnect_monitor_subscription(
-            device=self._config.device,
-            signal=self._config.signal,
-            source=config.source,
-            async_update=self.async_update,
-            async_signal_name=config.async_signal_name,
-        )
+        self._cleanup_data_api_subscription()
 
         # Reset async state
+        config = self.subscriptions["main"]
         self.async_update = False
         config.async_signal_name = None
         self._set_connection_status("disconnected")
@@ -755,131 +707,9 @@ class Image(ImageBase):
     # Image Update Methods
     ################################################################################
 
-    ########################################
-    # Connections
-
-    @SafeSlot(dict, dict)
-    def on_scan_status(self, msg: dict, meta: dict):
-        """
-        Initial scan status message handler, which is triggered at the beginning and end of scan.
-        Needed for setup of AsyncSignal connections.
-
-        Args:
-            msg(dict): The message content.
-            meta(dict): The message metadata.
-        """
-        current_scan_id = msg.get("scan_id", None)
-        if current_scan_id is None:
-            return
-        self._handle_scan_change(current_scan_id)
-
-    @SafeSlot(dict, dict)
-    def on_scan_progress(self, msg: dict, meta: dict):
-        """
-        For setting async image readback during scan progress updates if widget is started later than scan.
-
-        Args:
-            msg(dict): The message content.
-            meta(dict): The message metadata.
-        """
-        current_scan_id = meta.get("scan_id", None)
-        if current_scan_id is None:
-            return
-        self._handle_scan_change(current_scan_id)
-
-    def _handle_scan_change(self, current_scan_id: str):
-        """
-        Update internal scan ids and refresh async connections if needed.
-        Also clears image buffers when scan changes.
-
-        Args:
-            current_scan_id (str): The current scan identifier.
-        """
-        if current_scan_id == self.scan_id:
-            return
-
-        # Scan ID changed - clear buffers and reset image
-        self.old_scan_id = self.scan_id
-        self.scan_id = current_scan_id
-
-        # Clear image buffer for 1D data accumulation
-        self.main_image.clear()
-        if hasattr(self.main_image, "buffer"):
-            self.main_image.buffer = []
-            self.main_image.max_len = 0
-
-        # Reset crosshair if present
-        if self.crosshair is not None:
-            self.crosshair.reset()
-
-        # Reconnect async image subscription with new scan_id
-        if self.async_update:
-            self._setup_async_image(scan_id=self.scan_id)
-
-    def _get_async_signal_name(self) -> tuple[str, str] | None:
-        """
-        Returns device and async signal names used for endpoints/messages.
-
-        Returns:
-            tuple[str, str] | None: (device, async_signal_name) or None if not available.
-        """
-        if not self._config.device or not self._config.signal:
-            return None
-
-        config = self.subscriptions["main"]
-        async_signal = config.async_signal_name or self._config.signal
-        return self._config.device, async_signal
-
-    def _setup_async_image(self, scan_id: str | None):
-        """
-        (Re)connect async image readback for the current scan.
-
-        Args:
-            scan_id (str | None): The scan identifier to subscribe to.
-        """
-        if not self.async_update:
-            return
-
-        config = self.subscriptions["main"]
-        async_names = self._get_async_signal_name()
-        if async_names is None:
-            logger.info("Async image setup skipped because monitor information is incomplete.")
-            return
-
-        device, async_signal = async_names
-        if config.monitor_type == "1d":
-            slot = self.on_image_update_1d
-        elif config.monitor_type == "2d":
-            slot = self.on_image_update_2d
-        else:
-            logger.warning(
-                f"Async image setup skipped due to unsupported monitor type '{config.monitor_type}'."
-            )
-            return
-
-        # Disconnect any previous scan subscriptions to avoid stale updates.
-        for prev_scan_id in (self.old_scan_id, self.scan_id):
-            if prev_scan_id is None:
-                continue
-            self.bec_dispatcher.disconnect_slot(
-                slot, MessageEndpoints.device_async_signal(prev_scan_id, device, async_signal)
-            )
-
-        if scan_id is None:
-            logger.info("Scan ID not available yet; delaying async image subscription.")
-            return
-
-        self.bec_dispatcher.connect_slot(
-            slot,
-            MessageEndpoints.device_async_signal(scan_id, device, async_signal),
-            from_start=True,
-            cb_info={"scan_id": scan_id},
-        )
-        logger.info(f"Setup async image for {device}.{async_signal} and scan {scan_id}.")
-
     def disconnect_monitor(self, device: str | None = None, signal: str | None = None):
         """
-        Disconnect the monitor from the image update signals, both 1D and 2D.
+        Disconnect the monitor from the image update stream, both 1D and 2D.
 
         Args:
             device(str|None): The name of the device to disconnect. Defaults to current device.
@@ -899,109 +729,122 @@ class Image(ImageBase):
             )
             return
 
-        self._disconnect_monitor_subscription(
-            device=target_device,
-            signal=target_entry,
-            source=config.source,
-            async_update=self.async_update,
-            async_signal_name=config.async_signal_name,
-        )
-
-        self.subscriptions["main"].async_signal_name = None
-        self.async_update = False
+        self._disconnect_current_monitor()
         self._sync_device_selection()
 
-    ########################################
-    # 1D updates
-
-    @SafeSlot(dict, dict)
-    def on_image_update_1d(self, msg: dict, metadata: dict):
+    @SafeSlot(object)
+    def _on_data_update(self, update) -> None:
         """
-        Update the image with 1D data.
-        For preview signals: metadata doesn't contain scan_id.
-        For async signals: scan_id is managed via on_scan_status/on_scan_progress.
+        Render one columnar DataAPI update (live, backfill or history).
+
+        2-D sources display the latest frame; 1-D sources rebuild the
+        waterfall buffer from the columnar fragments (newest row last).
 
         Args:
-            msg(dict): The message containing the data.
-            metadata(dict): The metadata associated with the message.
+            update (SubscriptionUpdate): Full-state columnar snapshot.
         """
-        try:
-            image = self.main_image
-        except Exception:
+        if self._source_key is None:
             return
-        data = self._get_payload_data(msg)
-
+        source = update.sources.get(self._source_key)
+        if source is None or source.values is None or len(source.values) == 0:
+            return
+        self._handle_scan_rollover(update, source)
+        if self.subscriptions["main"].monitor_type == "2d":
+            data = np.asarray(source.values[-1])
+        else:
+            data = self._build_1d_buffer(source)
         if data is None:
-            logger.warning("No data received for image update from 1D.")
             return
-
-        image_buffer = self.adjust_image_buffer(image, data)
-
-        if self._color_bar is not None:
-            self._color_bar.blockSignals(True)
-        image.set_data(image_buffer)
-        if self._color_bar is not None:
-            self._color_bar.blockSignals(False)
-        if self._autorange_on_next_update:
-            self._autorange_on_next_update = False
-            self.auto_range()
-        self.image_updated.emit()
+        self._render_image_data(data)
 
     @staticmethod
-    def adjust_image_buffer(image: ImageItem, new_data: np.ndarray) -> np.ndarray:
+    def _effective_scan_id(update, source) -> str | None:
         """
-        Adjusts the image buffer to accommodate the new data, ensuring that all rows have the same length.
+        The scan id an update belongs to: the bound scan for scan-scoped
+        subscriptions, the last-seen scan id from the stream metadata for
+        scan-less preview streams.
 
         Args:
-            image: The image object (used to store a buffer list and max_len).
-            new_data (np.ndarray): The new incoming 1D waveform data.
+            update (SubscriptionUpdate): The update snapshot.
+            source (SourceData): The rendered source of the update.
 
         Returns:
-            np.ndarray: The updated image buffer with adjusted shapes.
+            str | None: The scan id, or None if not known (yet).
         """
-        # Guard for wrong data shapes
-        new_data = np.atleast_1d(np.asarray(new_data))
-        new_len = new_data.shape[0]
-        if not hasattr(image, "buffer"):
-            image.buffer = []
-            image.max_len = 0
+        if update.scan_id:
+            return update.scan_id
+        return source.metadata.get("scan_id")
 
-        if new_len > image.max_len:
-            image.max_len = new_len
-            for i in range(len(image.buffer)):
-                wf = image.buffer[i]
-                pad_width = image.max_len - wf.shape[0]
-                if pad_width > 0:
-                    image.buffer[i] = np.pad(wf, (0, pad_width), mode="constant", constant_values=0)
-            image.buffer.append(new_data)
-        else:
-            pad_width = image.max_len - new_len
-            if pad_width > 0:
-                new_data = np.pad(new_data, (0, pad_width), mode="constant", constant_values=0)
-            image.buffer.append(new_data)
-
-        image_buffer = np.array(image.buffer)
-        return image_buffer
-
-    ########################################
-    # 2D updates
-
-    @SafeSlot(dict, dict)
-    def on_image_update_2d(self, msg: dict, metadata: dict):
+    def _handle_scan_rollover(self, update, source) -> None:
         """
-        Update the image with 2D data.
+        Reset per-scan display state once the data belongs to a new scan.
+
+        Scan-scoped subscriptions deliver fresh per-scan series, so only the
+        bookkeeping and the crosshair need a reset. Scan-less preview streams
+        retain pre-rollover points; the display window is restricted to the
+        newest point (the one that carried the new scan id) onward.
 
         Args:
-            msg(dict): The message containing the data.
-            metadata(dict): The metadata associated with the message.
+            update (SubscriptionUpdate): The update snapshot.
+            source (SourceData): The rendered source of the update.
+        """
+        scan_id = self._effective_scan_id(update, source)
+        if scan_id is None or scan_id == self.scan_id:
+            return
+        previous = self.scan_id
+        self.old_scan_id = previous
+        self.scan_id = scan_id
+        if previous is None:
+            return
+        if source.kind == "unindexed" and source.ordinals:
+            self._min_display_ordinal = source.ordinals[-1]
+        if self.crosshair is not None:
+            self.crosshair.reset()
+
+    def _build_1d_buffer(self, source) -> np.ndarray | None:
+        """
+        Rebuild 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.
+
+        Args:
+            source (SourceData): The 1-D source snapshot.
+
+        Returns:
+            np.ndarray | None: The (n_rows, max_len) buffer, or None if no
+                displayable rows remain.
+        """
+        values = source.values
+        if self._min_display_ordinal is not None:
+            values = [
+                value
+                for ordinal, value in zip(source.ordinals, values)
+                if ordinal >= self._min_display_ordinal
+            ]
+        rows = [np.atleast_1d(np.asarray(value)) for value in values]
+        rows = [row for row in rows if row.ndim == 1]
+        if not rows:
+            return None
+        max_len = max(row.shape[0] for row in rows)
+        return np.array(
+            [
+                np.pad(row, (0, max_len - row.shape[0]), mode="constant", constant_values=0)
+                for row in rows
+            ]
+        )
+
+    def _render_image_data(self, data: np.ndarray) -> None:
+        """
+        Display the given data on the main image (shared render tail).
+
+        Args:
+            data (np.ndarray): The 2-D buffer or frame to render.
         """
         try:
             image = self.main_image
-        except Exception:
-            return
-        data = self._get_payload_data(msg)
-        if data is None:
-            logger.warning("No data received for image update from 2D.")
+        except Exception:  # pylint: disable=broad-except
             return
         if self._color_bar is not None:
             self._color_bar.blockSignals(True)
@@ -1013,22 +856,6 @@ class Image(ImageBase):
             self.auto_range()
         self.image_updated.emit()
 
-    def _get_payload_data(self, msg: dict) -> np.ndarray | None:
-        """
-        Extract payload from async/preview/monitor1D/2D message structures due to inconsistent formats in backend.
-
-        Args:
-            msg (dict): The incoming message containing data.
-        """
-        if not self.async_update:
-            return msg.get("data")
-        async_names = self._get_async_signal_name()
-        if async_names is None:
-            logger.warning("Async payload extraction failed; monitor info incomplete.")
-            return None
-        _, async_signal = async_names
-        return msg.get("signals", {}).get(async_signal, {}).get("value", None)
-
     ################################################################################
     # Clean up
     ################################################################################
@@ -1056,9 +883,8 @@ class Image(ImageBase):
         """
         self.layer_removed.disconnect(self._on_layer_removed)
 
-        # Disconnect current monitor
-        if self._config.device and self._config.signal:
-            self._disconnect_current_monitor()
+        # Close the DataAPI subscription
+        self._cleanup_data_api_subscription()
 
         self.subscriptions.clear()
 
diff --git a/tests/unit_tests/test_image_view_next_gen.py b/tests/unit_tests/test_image_view_next_gen.py
index 5e82c47d..0f892a4a 100644
--- a/tests/unit_tests/test_image_view_next_gen.py
+++ b/tests/unit_tests/test_image_view_next_gen.py
@@ -1,7 +1,9 @@
+from unittest.mock import MagicMock
+
 import numpy as np
 import pyqtgraph as pg
 import pytest
-from bec_lib.endpoints import MessageEndpoints
+from bec_lib.data_api.models import SourceData, SubscriptionUpdate
 from qtpy.QtCore import QPointF, Qt
 
 from bec_widgets.widgets.plots.image.bec_histogram_lut_item import (
@@ -30,6 +32,75 @@ def _set_signal_config(
     }
 
 
+def _fake_bridge_factory(monkeypatch):
+    """Replace the widget's QtDataSubscription with a recording fake."""
+    created = []
+
+    class _FakeBridge:
+        def __init__(
+            self, client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None
+        ):
+            self.client = client
+            self.sources = list(sources)
+            self.scan = scan
+            self.max_points = max_points
+            self.healthy = True
+            self.closed = False
+            self.updated = MagicMock()
+
+        def close(self):
+            self.closed = True
+
+    def factory(client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None):
+        bridge = _FakeBridge(client, sources, scan=scan, max_points=max_points)
+        created.append(bridge)
+        return bridge
+
+    monkeypatch.setattr("bec_widgets.widgets.plots.image.image.QtDataSubscription", factory)
+    return created
+
+
+def _make_source(
+    device, entry, values, *, kind="async", ordinals=None, metadata=None, as_numpy=False
+):
+    values = np.asarray(values) if as_numpy else tuple(values)
+    if ordinals is None:
+        ordinals = tuple(range(len(values)))
+    wrap = np.asarray if as_numpy else tuple
+    return SourceData(
+        device=device,
+        entry=entry,
+        kind=kind,
+        ordinals=wrap(ordinals),
+        values=values,
+        timestamps=wrap([float(i) for i in range(len(values))]),
+        complete=True,
+        metadata=metadata or {},
+    )
+
+
+def _make_update(source, scan_id="scan_1", reason="live"):
+    return SubscriptionUpdate(
+        scan_id=scan_id,
+        reason=reason,
+        sources={source.key: source},
+        aligned_ordinals=source.ordinals,
+        complete=True,
+        metadata={"group": "standalone"},
+    )
+
+
+def _preview_update(device, entry, values, *, ordinals=None, scan_id=None):
+    """Build an update as delivered by a scan-less device_preview stream."""
+    metadata = {"stream": "preview"}
+    if scan_id is not None:
+        metadata["scan_id"] = scan_id
+    source = _make_source(
+        device, entry, values, kind="unindexed", ordinals=ordinals, metadata=metadata
+    )
+    return _make_update(source, scan_id="")
+
+
 class _FakeClickEvent:
     """Minimal stand-in for a pyqtgraph MouseClickEvent."""
 
@@ -294,13 +365,12 @@ def test_enable_colorbar_with_vrange(qtbot, mocked_client, colorbar_type):
 # Device/signal update mechanism
 
 
-def test_image_setup_preview_signal_1d(qtbot, mocked_client):
+def test_image_setup_preview_signal_1d(qtbot, mocked_client, monkeypatch):
     """
-    Ensure that calling .image() with a 1‑D PreviewSignal connects using the 1‑D path
-    and updates correctly.
+    Ensure that calling .image() with a 1‑D PreviewSignal connects through a
+    scan-less DataAPI bridge and accumulates rows into the waterfall buffer.
     """
-    import numpy as np
-
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
 
     _set_signal_config(
@@ -321,20 +391,25 @@ def test_image_setup_preview_signal_1d(qtbot, mocked_client):
     assert view.device == "waveform1d"
     assert view.signal == "img"
 
-    # Simulate a waveform update from the dispatcher
+    # Preview signals are scan-less device streams with bounded retention
+    bridge = bridges[-1]
+    assert bridge.scan is None
+    assert bridge.sources == [("waveform1d", "img")]
+    assert bridge.max_points == Image.PREVIEW_1D_MAX_ROWS
+
+    # Simulate a waveform update from the DataAPI
     waveform = np.arange(25, dtype=float)
-    view.on_image_update_1d({"data": waveform}, {"scan_id": "scan_test"})
+    view._on_data_update(_preview_update("waveform1d", "img", [waveform]))
     assert view.main_image.raw_data.shape == (1, 25)
     np.testing.assert_array_equal(view.main_image.raw_data[0], waveform)
 
 
-def test_image_setup_preview_signal_2d(qtbot, mocked_client):
+def test_image_setup_preview_signal_2d(qtbot, mocked_client, monkeypatch):
     """
-    Ensure that calling .image() with a 2‑D PreviewSignal connects using the 2‑D path
-    and updates correctly.
+    Ensure that calling .image() with a 2‑D PreviewSignal connects through a
+    scan-less DataAPI bridge and displays the newest frame.
     """
-    import numpy as np
-
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
 
     _set_signal_config(
@@ -355,46 +430,40 @@ def test_image_setup_preview_signal_2d(qtbot, mocked_client):
     assert view.device == "eiger"
     assert view.signal == "img2d"
 
-    # Simulate a 2‑D image update
+    bridge = bridges[-1]
+    assert bridge.scan is None
+    assert bridge.sources == [("eiger", "img2d")]
+    assert bridge.max_points == Image.PREVIEW_2D_MAX_FRAMES
+
+    # Only the newest frame of the stream is displayed
     test_data = np.arange(16, dtype=float).reshape(4, 4)
-    view.on_image_update_2d({"data": test_data}, {})
+    view._on_data_update(_preview_update("eiger", "img2d", [np.zeros((4, 4)), test_data]))
     np.testing.assert_array_equal(view.main_image.image, test_data)
 
 
-def test_switching_device_disconnects_previous_preview_endpoint(qtbot, mocked_client, monkeypatch):
+def test_switching_device_replaces_preview_bridge(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2)
     _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=2)
 
-    connected = []
-    disconnected = []
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "connect_slot",
-        lambda slot, endpoint, *args, **kwargs: connected.append(endpoint),
-    )
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "disconnect_slot",
-        lambda slot, endpoint, *args, **kwargs: disconnected.append(endpoint),
-    )
-
     view.image(device="eiger", signal="img")
-    connected.clear()
-    disconnected.clear()
+    first = bridges[-1]
+    assert first.sources == [("eiger", "img")]
 
     view.device = "waveform1d"
 
-    assert MessageEndpoints.device_preview("eiger", "img") in disconnected
-    assert MessageEndpoints.device_preview("waveform1d", "img") in connected
+    assert first.closed is True
+    assert bridges[-1].sources == [("waveform1d", "img")]
+    assert bridges[-1].scan is None
 
 
-def test_switching_device_disconnects_previous_async_endpoint(qtbot, mocked_client, monkeypatch):
+def test_switching_device_replaces_async_bridge(qtbot, mocked_client, monkeypatch):
     """
-    Verify that switching device while async_update=True disconnects device_async_signal
-    endpoints for both scan_id and old_scan_id on the old device before reconnecting to
-    the new device.
+    Verify that switching device while async_update=True closes the previous
+    scan-scoped bridge before subscribing to the new device.
     """
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(
         mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj"
@@ -403,77 +472,44 @@ def test_switching_device_disconnects_previous_async_endpoint(qtbot, mocked_clie
         mocked_client, "waveform1d", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj"
     )
 
-    connected = []
-    disconnected = []
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "connect_slot",
-        lambda slot, endpoint, *args, **kwargs: connected.append(endpoint),
-    )
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "disconnect_slot",
-        lambda slot, endpoint, *args, **kwargs: disconnected.append(endpoint),
-    )
-
     view.image(device="eiger", signal="img")
     assert view.async_update is True
     assert view.subscriptions["main"].async_signal_name == "async_obj"
-
-    view.scan_id = "scan_current"
-    view.old_scan_id = "scan_previous"
-    connected.clear()
-    disconnected.clear()
+    first = bridges[-1]
+    assert first.scan == "live"
+    assert first.sources == [("eiger", "async_obj")]
 
     view.device = "waveform1d"
 
-    # Both scan_id and old_scan_id endpoints for the old device must be disconnected
-    assert (
-        MessageEndpoints.device_async_signal("scan_current", "eiger", "async_obj") in disconnected
-    )
-    assert (
-        MessageEndpoints.device_async_signal("scan_previous", "eiger", "async_obj") in disconnected
-    )
-    # The new device's async endpoint for the current scan must be connected
-    assert (
-        MessageEndpoints.device_async_signal("scan_current", "waveform1d", "async_obj") in connected
-    )
+    assert first.closed is True
+    assert bridges[-1].scan == "live"
+    assert bridges[-1].sources == [("waveform1d", "async_obj")]
 
 
-def test_switching_signal_disconnects_previous_preview_endpoint(qtbot, mocked_client, monkeypatch):
+def test_switching_signal_replaces_preview_bridge(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(mocked_client, "eiger", "img_a", signal_class="PreviewSignal", ndim=2)
     _set_signal_config(mocked_client, "eiger", "img_b", signal_class="PreviewSignal", ndim=2)
 
-    connected = []
-    disconnected = []
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "connect_slot",
-        lambda slot, endpoint, *args, **kwargs: connected.append(endpoint),
-    )
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "disconnect_slot",
-        lambda slot, endpoint, *args, **kwargs: disconnected.append(endpoint),
-    )
-
     view.image(device="eiger", signal="img_a")
-    connected.clear()
-    disconnected.clear()
+    first = bridges[-1]
+    assert first.sources == [("eiger", "img_a")]
 
     view.signal = "img_b"
 
-    assert MessageEndpoints.device_preview("eiger", "img_a") in disconnected
-    assert MessageEndpoints.device_preview("eiger", "img_b") in connected
+    assert first.closed is True
+    assert bridges[-1].sources == [("eiger", "img_b")]
+    assert bridges[-1].scan is None
 
 
-def test_switching_signal_disconnects_previous_async_endpoint(qtbot, mocked_client, monkeypatch):
+def test_switching_signal_replaces_async_bridge(qtbot, mocked_client, monkeypatch):
     """
-    When the current monitor is an async signal, switching to a different signal must
-    disconnect the previous async endpoint (based on scan_id/async_signal_name) before
-    reconnecting with the new signal's async endpoint.
+    When the current monitor is an async signal, switching to a different
+    signal must close the previous bridge and subscribe with the new signal's
+    obj_name as the DataAPI entry.
     """
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(
         mocked_client, "eiger", "img_a", signal_class="AsyncSignal", ndim=2, obj_name="async_obj_a"
@@ -482,40 +518,20 @@ def test_switching_signal_disconnects_previous_async_endpoint(qtbot, mocked_clie
         mocked_client, "eiger", "img_b", signal_class="AsyncSignal", ndim=2, obj_name="async_obj_b"
     )
 
-    connected = []
-    disconnected = []
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "connect_slot",
-        lambda slot, endpoint, *args, **kwargs: connected.append(endpoint),
-    )
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "disconnect_slot",
-        lambda slot, endpoint, *args, **kwargs: disconnected.append(endpoint),
-    )
-
-    # Connect to img_a as an async signal; scan_id is None so no actual subscription is made
     view.image(device="eiger", signal="img_a")
     assert view.async_update is True
     assert view.subscriptions["main"].async_signal_name == "async_obj_a"
     assert view.subscriptions["main"].source == "device_monitor_2d"
-
-    # Simulate an active scan so that the async endpoint is real
-    view.scan_id = "scan_123"
-    connected.clear()
-    disconnected.clear()
+    first = bridges[-1]
+    assert first.sources == [("eiger", "async_obj_a")]
 
     # Switch to a different signal
     view.signal = "img_b"
 
-    # The previous async endpoint for img_a must have been disconnected
-    expected_disconnect = MessageEndpoints.device_async_signal("scan_123", "eiger", "async_obj_a")
-    assert expected_disconnect in disconnected
-
-    # The new async endpoint for img_b must have been connected
-    expected_connect = MessageEndpoints.device_async_signal("scan_123", "eiger", "async_obj_b")
-    assert expected_connect in connected
+    assert first.closed is True
+    assert view.subscriptions["main"].async_signal_name == "async_obj_b"
+    assert bridges[-1].sources == [("eiger", "async_obj_b")]
+    assert bridges[-1].scan == "live"
 
 
 def test_preview_signals_skip_0d_entries(qtbot, mocked_client, monkeypatch):
@@ -568,8 +584,10 @@ def test_preview_signals_skip_0d_entries(qtbot, mocked_client, monkeypatch):
 
 def test_image_async_signal_uses_obj_name(qtbot, mocked_client, monkeypatch):
     """
-    Verify async signals use obj_name for endpoints/payloads and reconnect with scan_id.
+    Verify async signals subscribe with their obj_name as the DataAPI entry
+    and render the delivered fragments.
     """
+    bridges = _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"
@@ -579,53 +597,33 @@ def test_image_async_signal_uses_obj_name(qtbot, mocked_client, monkeypatch):
     assert view.subscriptions["main"].async_signal_name == "async_obj"
     assert view.async_update is True
 
-    # Prepare scan ids and capture dispatcher calls
-    view.old_scan_id = "old_scan"
-    view.scan_id = "new_scan"
-    connected = []
-    disconnected = []
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "connect_slot",
-        lambda slot, endpoint, from_start=False, cb_info=None: connected.append(
-            (slot, endpoint, from_start, cb_info)
-        ),
-    )
-    monkeypatch.setattr(
-        view.bec_dispatcher,
-        "disconnect_slot",
-        lambda slot, endpoint: disconnected.append((slot, endpoint)),
-    )
+    bridge = bridges[-1]
+    assert bridge.scan == "live"
+    assert bridge.sources == [("eiger", "async_obj")]
+    assert bridge.max_points is None
 
-    view._setup_async_image(view.scan_id)
-
-    expected_new = MessageEndpoints.device_async_signal("new_scan", "eiger", "async_obj")
-    expected_old = MessageEndpoints.device_async_signal("old_scan", "eiger", "async_obj")
-    assert any(ep == expected_new for _, ep, _, _ in connected)
-    assert any(ep == expected_old for _, ep in disconnected)
-
-    # Payload extraction should use obj_name
+    # Rendering reads the source keyed by (device, obj_name)
     payload = np.array([1, 2, 3])
-    msg = {"signals": {"async_obj": {"value": payload}}}
-    assert np.array_equal(view._get_payload_data(msg), payload)
+    source = _make_source(
+        "eiger", "async_obj", [payload], metadata={"async_update_type": "add", "max_shape": [None]}
+    )
+    view._on_data_update(_make_update(source))
+    np.testing.assert_array_equal(view.main_image.raw_data, payload.reshape(1, 3))
 
 
 def test_disconnect_clears_async_state(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(
         mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj"
     )
 
     view.image(device="eiger", signal="img")
-    view.scan_id = "scan_x"
-    view.old_scan_id = "scan_y"
-    view.subscriptions["main"].async_signal_name = "async_obj"
-
-    # Avoid touching real dispatcher
-    monkeypatch.setattr(view.bec_dispatcher, "disconnect_slot", lambda *args, **kwargs: None)
+    assert view.async_update is True
 
     view.disconnect_monitor(device="eiger", signal="img")
 
+    assert bridges[-1].closed is True
     assert view.subscriptions["main"].async_signal_name is None
     assert view.async_update is False
 
@@ -634,7 +632,8 @@ def test_disconnect_clears_async_state(qtbot, mocked_client, monkeypatch):
 # Connection guardrails
 
 
-def test_image_setup_rejects_unsupported_signal_class(qtbot, mocked_client):
+def test_image_setup_rejects_unsupported_signal_class(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(mocked_client, "eiger", "img", signal_class="Signal", ndim=2)
 
@@ -643,9 +642,11 @@ def test_image_setup_rejects_unsupported_signal_class(qtbot, mocked_client):
     assert view.subscriptions["main"].source is None
     assert view.subscriptions["main"].monitor_type is None
     assert view.async_update is False
+    assert bridges == []
 
 
-def test_image_disconnects_with_missing_entry(qtbot, mocked_client):
+def test_image_disconnects_with_missing_entry(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
     _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2)
 
@@ -656,85 +657,163 @@ def test_image_disconnects_with_missing_entry(qtbot, mocked_client):
     view.image(device="eiger", signal=None)
     assert view.device == ""
     assert view.signal == ""
+    assert all(bridge.closed for bridge in bridges)
 
 
-def test_handle_scan_change_clears_buffers_and_resets_crosshair(qtbot, mocked_client, monkeypatch):
+def test_scan_rollover_resets_accumulation_and_crosshair(qtbot, mocked_client, monkeypatch):
+    """A new scan id in the update stream starts a fresh 1D accumulation and
+    resets the crosshair (replaces the legacy _handle_scan_change data path)."""
+    _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
-    view.scan_id = "scan_1"
-    view.main_image.buffer = [np.array([1.0, 2.0])]
-    view.main_image.max_len = 2
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj"
+    )
+    view.image(device="eiger", signal="img")
 
-    clear_called = []
-    monkeypatch.setattr(view.main_image, "clear", lambda: clear_called.append(True))
+    view._on_data_update(
+        _make_update(_make_source("eiger", "async_obj", [np.arange(4)]), scan_id="scan_1")
+    )
+    assert view.scan_id == "scan_1"
+    assert view.main_image.raw_data.shape == (1, 4)
+
+    view.hook_crosshair()
     reset_called = []
-    if view.crosshair is not None:
-        monkeypatch.setattr(view.crosshair, "reset", lambda: reset_called.append(True))
+    monkeypatch.setattr(view.crosshair, "reset", lambda: reset_called.append(True))
 
-    view._handle_scan_change("scan_2")
+    # First update of the new scan: the DataAPI delivers a fresh per-scan series
+    view._on_data_update(
+        _make_update(_make_source("eiger", "async_obj", [np.arange(6)]), scan_id="scan_2")
+    )
 
     assert view.old_scan_id == "scan_1"
     assert view.scan_id == "scan_2"
-    assert clear_called == [True]
-    assert view.main_image.buffer == []
-    assert view.main_image.max_len == 0
-    if view.crosshair is not None:
-        assert reset_called == [True]
+    assert reset_called == [True]
+    assert view.main_image.raw_data.shape == (1, 6)
 
 
-def test_handle_scan_change_reconnects_async(qtbot, mocked_client, monkeypatch):
+def test_preview_scan_rollover_restricts_display_window(qtbot, mocked_client, monkeypatch):
+    """Preview streams retain pre-rollover points; a scan change restricts the
+    displayed accumulation to the newest point onward."""
+    _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
-    view.scan_id = "scan_1"
-    view.async_update = True
+    _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1)
+    view.image(device="waveform1d", signal="img")
 
-    called = []
-    monkeypatch.setattr(view, "_setup_async_image", lambda scan_id: called.append(scan_id))
+    rows = [np.full(5, i, dtype=float) for i in range(3)]
+    view._on_data_update(_preview_update("waveform1d", "img", rows, scan_id="scan_1"))
+    assert view.scan_id == "scan_1"
+    assert view.main_image.raw_data.shape == (3, 5)
 
-    view._handle_scan_change("scan_2")
+    # The device stream keeps the old rows; the newest one belongs to scan_2.
+    rows_after = rows + [np.full(5, 99.0)]
+    view._on_data_update(_preview_update("waveform1d", "img", rows_after, scan_id="scan_2"))
 
-    assert called == ["scan_2"]
+    assert view.old_scan_id == "scan_1"
+    assert view.scan_id == "scan_2"
+    assert view.main_image.raw_data.shape == (1, 5)
+    np.testing.assert_array_equal(view.main_image.raw_data[0], np.full(5, 99.0))
+
+    # Subsequent scan_2 rows extend the new accumulation.
+    rows_more = rows_after + [np.full(5, 100.0)]
+    view._on_data_update(_preview_update("waveform1d", "img", rows_more, scan_id="scan_2"))
+    assert view.main_image.raw_data.shape == (2, 5)
 
 
-def test_handle_scan_change_same_scan_noop(qtbot, mocked_client, monkeypatch):
+def test_scan_rollover_same_scan_noop(qtbot, mocked_client, monkeypatch):
+    _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
-    view.scan_id = "scan_1"
-    view.main_image.buffer = [np.array([1.0])]
-    view.main_image.max_len = 1
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj"
+    )
+    view.image(device="eiger", signal="img")
 
-    clear_called = []
-    monkeypatch.setattr(view.main_image, "clear", lambda: clear_called.append(True))
-
-    view._handle_scan_change("scan_1")
+    view._on_data_update(
+        _make_update(_make_source("eiger", "async_obj", [np.arange(3)]), scan_id="scan_1")
+    )
+    view._on_data_update(
+        _make_update(
+            _make_source("eiger", "async_obj", [np.arange(3), np.arange(3)]), scan_id="scan_1"
+        )
+    )
 
     assert view.scan_id == "scan_1"
-    assert clear_called == []
-    assert view.main_image.buffer == [np.array([1.0])]
-    assert view.main_image.max_len == 1
+    assert view.old_scan_id is None
+    assert view.main_image.raw_data.shape == (2, 3)
 
 
-def test_image_data_update_2d(qtbot, mocked_client):
+def test_image_data_update_2d(qtbot, mocked_client, monkeypatch):
+    _fake_bridge_factory(monkeypatch)
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    test_data = np.random.rand(20, 30)
-    message = {"data": test_data}
-    metadata = {}
+    _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2)
+    bec_image_view.image(device="eiger", signal="img")
 
-    bec_image_view.on_image_update_2d(message, metadata)
+    test_data = np.random.rand(20, 30)
+    bec_image_view._on_data_update(_preview_update("eiger", "img", [test_data]))
 
     np.testing.assert_array_equal(bec_image_view.main_image.image, test_data)
 
 
-def test_image_data_update_1d(qtbot, mocked_client):
+def test_image_data_update_1d(qtbot, mocked_client, monkeypatch):
+    _fake_bridge_factory(monkeypatch)
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
+    _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1)
+    bec_image_view.image(device="waveform1d", signal="img")
+
     waveform1 = np.random.rand(50)
     waveform2 = np.random.rand(60)  # Different length, tests padding logic
-    metadata = {"scan_id": "scan_test"}
 
-    bec_image_view.on_image_update_1d({"data": waveform1}, metadata)
+    bec_image_view._on_data_update(_preview_update("waveform1d", "img", [waveform1]))
     assert bec_image_view.main_image.raw_data.shape == (1, 50)
 
-    bec_image_view.on_image_update_1d({"data": waveform2}, metadata)
+    bec_image_view._on_data_update(_preview_update("waveform1d", "img", [waveform1, waveform2]))
     assert bec_image_view.main_image.raw_data.shape == (2, 60)
 
 
+def test_async_add_slice_displays_accumulated_rows(qtbot, mocked_client, monkeypatch):
+    """add_slice sources deliver one accumulated row per ordinal; shorter rows
+    are zero-padded to the longest one."""
+    _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")
+
+    source = _make_source(
+        "eiger",
+        "async_obj",
+        [[1.0, 2.0, 3.0], [4.0]],
+        metadata={"async_update_type": "add_slice", "max_shape": [None, None]},
+    )
+    view._on_data_update(_make_update(source))
+
+    np.testing.assert_array_equal(
+        view.main_image.raw_data, np.array([[1.0, 2.0, 3.0], [4.0, 0.0, 0.0]])
+    )
+
+
+def test_async_replace_displays_current_state(qtbot, mocked_client, monkeypatch):
+    """replace sources expose a single element: the current full state."""
+    _fake_bridge_factory(monkeypatch)
+    view = create_widget(qtbot, Image, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj"
+    )
+    view.image(device="eiger", signal="img")
+
+    frame = np.arange(12, dtype=float).reshape(3, 4)
+    source = _make_source(
+        "eiger",
+        "async_obj",
+        [frame],
+        ordinals=(0,),
+        metadata={"async_update_type": "replace", "max_shape": [3, 4]},
+    )
+    view._on_data_update(_make_update(source))
+
+    np.testing.assert_array_equal(view.main_image.image, frame)
+
+
 ##############################################
 # Toolbar and Actions Tests
 
@@ -826,6 +905,7 @@ def test_setting_vrange_with_colorbar(qtbot, mocked_client, colorbar_type):
 
 
 def test_setup_image_from_toolbar(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
 
     _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2)
@@ -1134,7 +1214,7 @@ def test_roi_plot_data_from_image(qtbot, mocked_client):
 
     # Provide deterministic 2D data
     test_data = np.arange(25).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
 
     # Activate ROI crosshair
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
@@ -1164,7 +1244,7 @@ def test_roi_plots_ignore_rgb_images_and_clear_stale_curves(qtbot, mocked_client
     """RGB images have vector-valued pixels and cannot produce scalar profile curves."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     scalar_image = np.arange(25).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": scalar_image}, {})
+    bec_image_view._render_image_data(scalar_image)
 
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
@@ -1177,7 +1257,7 @@ def test_roi_plots_ignore_rgb_images_and_clear_stale_curves(qtbot, mocked_client
     assert bec_image_view.y_roi_pinned is not None
 
     rgb_image = np.zeros((5, 5, 3), dtype=np.uint8)
-    bec_image_view.on_image_update_2d({"data": rgb_image}, {})
+    bec_image_view._render_image_data(rgb_image)
 
     assert bec_image_view._compute_image_slices(bec_image_view.main_image, 2, 3) is None
     assert bec_image_view.x_roi_curve is None
@@ -1194,7 +1274,7 @@ def test_pinned_roi_profiles_freeze_and_clear(qtbot, mocked_client):
 
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
 
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
@@ -1231,7 +1311,7 @@ def test_pinned_roi_profiles_keep_style_on_theme_change(qtbot, mocked_client):
     import numpy as np
 
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1253,7 +1333,7 @@ def test_pin_survives_scan_reset(qtbot, mocked_client):
     import numpy as np
 
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1263,7 +1343,7 @@ def test_pin_survives_scan_reset(qtbot, mocked_client):
     pin_point = bec_image_view.crosshair.pinned_point
     assert pin_point is not None
 
-    # Image._handle_scan_change calls crosshair.reset() on each new scan id.
+    # Image._handle_scan_rollover calls crosshair.reset() on each new scan id.
     bec_image_view.crosshair.reset()
 
     # The pin marker and the frozen reference profiles must still be there.
@@ -1279,7 +1359,7 @@ def test_pin_and_profiles_restored_after_roi_toggle(qtbot, mocked_client):
 
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1311,7 +1391,7 @@ def test_detached_pin_can_be_removed_with_right_click(qtbot, mocked_client, monk
     from qtpy.QtWidgets import QMenu
 
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1341,7 +1421,7 @@ def test_pinned_profiles_follow_image_updates(qtbot, mocked_client):
     exactly like the pin's intensity label."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25, dtype=float).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1351,7 +1431,7 @@ def test_pinned_profiles_follow_image_updates(qtbot, mocked_client):
     np.testing.assert_array_equal(x_pinned, test_data[:, 3])
 
     updated = test_data + 100.0
-    bec_image_view.on_image_update_2d({"data": updated}, {})
+    bec_image_view._render_image_data(updated)
 
     qtbot.waitUntil(
         lambda: np.array_equal(bec_image_view.x_roi_pinned.getData()[1], updated[:, 3]), timeout=500
@@ -1364,7 +1444,7 @@ def test_detached_pin_profiles_follow_image_updates(qtbot, mocked_client):
     """Pinned profiles keep following image data while the crosshair is toggled off."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25, dtype=float).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1375,7 +1455,7 @@ def test_detached_pin_profiles_follow_image_updates(qtbot, mocked_client):
     assert bec_image_view.x_roi_pinned is not None
 
     updated = test_data + 100.0
-    bec_image_view.on_image_update_2d({"data": updated}, {})
+    bec_image_view._render_image_data(updated)
 
     qtbot.waitUntil(
         lambda: np.array_equal(bec_image_view.x_roi_pinned.getData()[1], updated[:, 3]), timeout=500
@@ -1387,7 +1467,7 @@ def test_crosshair_moves_update_profiles_immediately(qtbot, mocked_client):
     them at the current crosshair position."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25, dtype=float).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
     switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair")
     switch.actions["crosshair_roi"].action.trigger()
     qtbot.wait(50)
@@ -1402,7 +1482,7 @@ def test_crosshair_moves_update_profiles_immediately(qtbot, mocked_client):
 
     # New frames refresh the live profile at the crosshair position.
     updated = test_data + 100.0
-    bec_image_view.on_image_update_2d({"data": updated}, {})
+    bec_image_view._render_image_data(updated)
     np.testing.assert_array_equal(bec_image_view.x_roi_curve.getData()[1], updated[:, 2])
 
 
@@ -1411,7 +1491,7 @@ def test_live_label_intensity_updates_on_image_update(qtbot, mocked_client):
     not only when the mouse moves."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
     test_data = np.arange(25, dtype=float).reshape(5, 5)
-    bec_image_view.on_image_update_2d({"data": test_data}, {})
+    bec_image_view._render_image_data(test_data)
     bec_image_view.hook_crosshair()
     bec_image_view.crosshair.plot_item.vb.setRange(
         xRange=(0, test_data.shape[0]), yRange=(0, test_data.shape[1]), padding=0
@@ -1420,14 +1500,14 @@ def test_live_label_intensity_updates_on_image_update(qtbot, mocked_client):
     bec_image_view.crosshair.mouse_moved(manual_pos=(2.5, 3.5))
     assert "Intensity: 13.000" in bec_image_view.crosshair.coord_label.toPlainText()
 
-    bec_image_view.on_image_update_2d({"data": test_data + 100.0}, {})
+    bec_image_view._render_image_data(test_data + 100.0)
     assert "Intensity: 113.000" in bec_image_view.crosshair.coord_label.toPlainText()
 
 
 def test_active_pin_label_intensity_updates_on_image_update(qtbot, mocked_client):
     """Pinned crosshair label intensity follows image updates at the pinned position."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     bec_image_view.hook_crosshair()
 
     bec_image_view.crosshair.set_pin(2.0, 3.0)
@@ -1435,7 +1515,7 @@ def test_active_pin_label_intensity_updates_on_image_update(qtbot, mocked_client
         "pin (2.500, 3.500)\nIntensity: 13.000"
     )
 
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5) + 100}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 100)
 
     qtbot.waitUntil(
         lambda: bec_image_view.crosshair.pinned_label.toPlainText()
@@ -1449,7 +1529,7 @@ def test_image_update_does_not_replay_active_crosshair_mouse_handling(
 ):
     """Image updates refresh a pin without re-snapping or emitting live crosshair updates."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     bec_image_view.hook_crosshair()
     bec_image_view.crosshair.set_pin(2.0, 3.0)
 
@@ -1460,7 +1540,7 @@ def test_image_update_does_not_replay_active_crosshair_mouse_handling(
         lambda *args, **kwargs: mouse_moves.append((args, kwargs)),
     )
 
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5) + 100}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 100)
 
     qtbot.waitUntil(
         lambda: bec_image_view.crosshair.pinned_label.toPlainText()
@@ -1473,7 +1553,7 @@ def test_image_update_does_not_replay_active_crosshair_mouse_handling(
 def test_detached_pin_label_intensity_updates_on_image_update(qtbot, mocked_client):
     """Detached pin labels keep following image updates while crosshair is disabled."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5)}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5))
     bec_image_view.hook_crosshair()
     bec_image_view.crosshair.set_pin(2.0, 3.0)
     bec_image_view.unhook_crosshair()
@@ -1484,7 +1564,7 @@ def test_detached_pin_label_intensity_updates_on_image_update(qtbot, mocked_clie
         "pin (2.500, 3.500)\nIntensity: 13.000"
     )
 
-    bec_image_view.on_image_update_2d({"data": np.arange(25).reshape(5, 5) + 200}, {})
+    bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 200)
 
     qtbot.waitUntil(
         lambda: bec_image_view._detached_pin["label"].toPlainText()
@@ -1496,7 +1576,7 @@ def test_detached_pin_label_intensity_updates_on_image_update(qtbot, mocked_clie
 def test_detached_pin_label_omits_scalar_intensity_for_rgb_image(qtbot, mocked_client):
     """Detached RGB pin labels only show coordinates because a pixel is an array."""
     bec_image_view = create_widget(qtbot, Image, client=mocked_client)
-    bec_image_view.on_image_update_2d({"data": np.zeros((5, 5, 3))}, {})
+    bec_image_view._render_image_data(np.zeros((5, 5, 3)))
     bec_image_view.hook_crosshair()
     bec_image_view.crosshair.set_pin(2.0, 3.0)
     bec_image_view.unhook_crosshair()
@@ -1588,7 +1668,7 @@ def test_log_scale_does_not_crash_with_full_colorbar(qtbot, mocked_client, data)
 
     view.log = True
     # This is the exact path that used to raise "Cannot set range [nan, nan]".
-    view.on_image_update_2d({"data": data}, {})
+    view._render_image_data(data)
 
     vmin, vmax = view.main_image.v_range
     assert np.isfinite(vmin) and np.isfinite(vmax)
@@ -1604,7 +1684,7 @@ def test_log_scale_all_negative_keeps_finite_image(qtbot, mocked_client):
     view.log = True
     data = (-np.abs(np.random.rand(15, 15)) - 1.0).astype(np.float32)
 
-    view.on_image_update_2d({"data": data}, {})
+    view._render_image_data(data)
 
     assert view.main_image.image is not None
     assert np.all(np.isfinite(view.main_image.image))
@@ -1613,7 +1693,7 @@ def test_log_scale_all_negative_keeps_finite_image(qtbot, mocked_client):
 def test_set_v_range_ignores_non_finite_levels(qtbot, mocked_client):
     """Non-finite v_range requests are rejected rather than forwarded to pg."""
     view = create_widget(qtbot, Image, client=mocked_client)
-    view.on_image_update_2d({"data": np.random.rand(10, 10)}, {})
+    view._render_image_data(np.random.rand(10, 10))
     good = view.main_image.v_range
 
     view.main_image.set_v_range((np.nan, np.nan))
@@ -1643,16 +1723,20 @@ def test_image_processor_log_is_finite_for_non_positive():
 ##############################################
 
 
-def test_adjust_image_buffer_coerces_list_and_scalar(qtbot, mocked_client):
+def test_1d_buffer_coerces_list_and_scalar(qtbot, mocked_client, monkeypatch):
     """Non-ndarray payloads (python list, 0-d scalar) must not crash the
     1D buffer accumulation."""
+    _fake_bridge_factory(monkeypatch)
     view = create_widget(qtbot, Image, client=mocked_client)
-    image = view.main_image
+    _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1)
+    view.image(device="waveform1d", signal="img")
 
-    buf = view.adjust_image_buffer(image, [1, 2, 3])  # python list
-    assert buf.shape == (1, 3)
-    buf = view.adjust_image_buffer(image, np.array(42.0))  # 0-d scalar
-    assert buf.shape[0] == 2
+    view._on_data_update(_preview_update("waveform1d", "img", [[1, 2, 3]]))  # python list
+    assert view.main_image.raw_data.shape == (1, 3)
+    view._on_data_update(
+        _preview_update("waveform1d", "img", [[1, 2, 3], np.array(42.0)])  # 0-d scalar
+    )
+    assert view.main_image.raw_data.shape == (2, 3)
 
 
 ##############################################
@@ -1675,3 +1759,25 @@ def test_layer_accessors_safe_after_teardown(qtbot, mocked_client):
     view.autorange_mode = "max"
     view._sync_autorange_switch()
     view._sync_colorbar_levels()
+
+
+def test_history_numpy_source_renders(qtbot, mocked_client, monkeypatch):
+    """Regression: numpy-array source columns from the history bulk path must
+    render (a truth-test on source.values raised ValueError, swallowed by
+    SafeSlot, leaving the image blank)."""
+    _fake_bridge_factory(monkeypatch)
+    view = create_widget(qtbot, Image, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj"
+    )
+    view.image(device="eiger", signal="img")
+    frame = np.arange(12, dtype=float).reshape(3, 4)
+    source = _make_source(
+        "eiger",
+        "async_obj",
+        np.asarray([frame]),
+        metadata={"async_update_type": "add", "max_shape": [None, None]},
+        as_numpy=True,
+    )
+    view._on_data_update(_make_update(source, reason="history"))
+    np.testing.assert_array_equal(view.main_image.raw_data, frame)