feat(multi_waveform): select device and signal with class-based routing

bec_widgets/cli/client.py

@@ -4536,14 +4536,28 @@ class MultiWaveform(RPCBase):
     @rpc_call
     def monitor(self) -> "str":
         """
-        The monitor of the figure widget.
+        The monitored device of the figure widget (alias of ``device``).
         """
 
     @monitor.setter
     @rpc_call
     def monitor(self) -> "str":
         """
-        The monitor of the figure widget.
+        The monitored device of the figure widget (alias of ``device``).
+        """
+
+    @property
+    @rpc_call
+    def monitor_signal(self) -> "str":
+        """
+        The monitored signal/entry on the device (alias of ``signal``).
+        """
+
+    @monitor_signal.setter
+    @rpc_call
+    def monitor_signal(self) -> "str":
+        """
+        The monitored signal/entry on the device (alias of ``signal``).
         """
 
     @rpc_call
@@ -4557,11 +4571,18 @@ class MultiWaveform(RPCBase):
         """
 
     @rpc_call
-    def plot(self, monitor: "str", color_palette: "str | None" = "plasma"):
+    def plot(
+        self, monitor: "str", signal: "str | None" = None, color_palette: "str | None" = "plasma"
+    ):
         """
-        Create a plot for the given monitor.
+        Create a plot for the given monitor device and signal.
+
         Args:
-            monitor (str): The monitor to set.
+            monitor (str): The device to monitor.
+            signal (str|None): The signal/entry to monitor on the device. The reserved
+                entry ``"monitor_1d"`` selects the scan-less device_monitor_1d stream.
+                If None, the signal is auto-selected: the device's only 1D-capable
+                preview/async signal if unambiguous, the monitor_1d stream otherwise.
             color_palette (str|None): The color palette to use for the plot.
         """
 

bec_widgets/widgets/control/device_input/device_combobox/device_combobox.py

@@ -44,6 +44,9 @@ class DeviceInputConfig(ConnectionConfig):
             editable-combobox completer.
         include_signals_with_write_access: Whether to additionally list signals that report
             write access, in addition to the devices selected by the other filters.
+        include_async_readout_devices: Whether to additionally list devices with async
+            readout priority (potential ``device_monitor_1d`` publishers), in addition to
+            the devices selected by the other filters.
     """
 
     device_filter: list[str] = Field(default_factory=list)
@@ -55,6 +58,7 @@ class DeviceInputConfig(ConnectionConfig):
     signal_class_filter: list[str] = Field(default_factory=list)
     autocomplete: bool = False
     include_signals_with_write_access: bool = False
+    include_async_readout_devices: bool = False
 
     @field_validator("device_filter")
     @classmethod
@@ -120,6 +124,9 @@ class DeviceComboBox(BECWidget, QComboBox):
             Qt's default editable-combobox completion behavior.
         include_signals_with_write_access: If True, additionally list signals that report
             write access on top of the devices selected by the other filters.
+        include_async_readout_devices: If True, additionally list devices with async readout
+            priority (potential ``device_monitor_1d`` publishers) on top of the devices
+            selected by the other filters.
         **kwargs: Additional keyword arguments passed to ``BECWidget``.
     """
 
@@ -152,6 +159,7 @@ class DeviceComboBox(BECWidget, QComboBox):
         signal_class_filter: list[str] | None = None,
         autocomplete: bool | None = None,
         include_signals_with_write_access: bool | None = None,
+        include_async_readout_devices: bool | None = None,
         **kwargs,
     ):
         self.config = self._process_config(config)
@@ -191,6 +199,8 @@ class DeviceComboBox(BECWidget, QComboBox):
             self.autocomplete = True
         if include_signals_with_write_access is not None:
             self.config.include_signals_with_write_access = include_signals_with_write_access
+        if include_async_readout_devices is not None:
+            self.config.include_async_readout_devices = include_async_readout_devices
 
         if available_devices is not None:
             self.set_available_devices(available_devices)
@@ -273,6 +283,14 @@ class DeviceComboBox(BECWidget, QComboBox):
                 if name not in seen:
                     device_names.append(name)
                     seen.add(name)
+        if self.include_async_readout_devices:
+            seen = set(device_names)
+            for device in enabled_devices:
+                if device.name in seen:
+                    continue
+                if getattr(device, "readout_priority", None) == ReadoutPriority.ASYNC:
+                    device_names.append(device.name)
+                    seen.add(device.name)
         self.devices = device_names
 
     @SafeSlot(list)
@@ -336,6 +354,16 @@ class DeviceComboBox(BECWidget, QComboBox):
         self.config.include_signals_with_write_access = value
         self.update_devices_from_filters()
 
+    @SafeProperty(bool)
+    def include_async_readout_devices(self) -> bool:
+        """Whether async-readout devices are listed in addition to the other filters."""
+        return self.config.include_async_readout_devices
+
+    @include_async_readout_devices.setter
+    def include_async_readout_devices(self, value: bool):
+        self.config.include_async_readout_devices = value
+        self.update_devices_from_filters()
+
     @SafeProperty(bool)
     def filter_to_device(self):
         """Include generic Device objects."""

bec_widgets/widgets/control/device_input/signal_combobox/signal_combobox.py

@@ -66,6 +66,8 @@ class SignalComboBox(BECWidget, QComboBox):
         require_device: If True, class-based signal filtering requires a valid selected device.
         autocomplete: If True, use the explicit line-edit style completer. If False, keep
             Qt's default editable-combobox completion behavior.
+        extra_class_signals: Static ``(display_name, signal_config)`` entries prepended to
+            class-based signal lists, e.g. reserved stream names such as ``"monitor_1d"``.
         **kwargs: Additional keyword arguments passed to ``BECWidget``.
     """
 
@@ -93,6 +95,7 @@ class SignalComboBox(BECWidget, QComboBox):
         store_signal_config: bool = True,
         require_device: bool = False,
         autocomplete: bool | None = None,
+        extra_class_signals: list[tuple[str, dict]] | None = None,
         **kwargs,
     ):
         self.config = self._process_config(config)
@@ -109,6 +112,7 @@ class SignalComboBox(BECWidget, QComboBox):
         self._signal_class_filter = signal_class_filter or []
         self._store_signal_config = store_signal_config
         self._require_device = require_device
+        self._extra_class_signals = list(extra_class_signals or [])
         self._is_valid_input = False
         self._completer_model = QStringListModel(self)
 
@@ -500,7 +504,12 @@ class SignalComboBox(BECWidget, QComboBox):
         """
         if not self._store_signal_config:
             return None
-        signal_info = self.itemData(self.currentIndex())
+        # Resolve by the displayed text first: on editable comboboxes the current
+        # index does not follow programmatic text changes.
+        index = self._find_signal_index(self.currentText())
+        if index < 0:
+            index = self.currentIndex()
+        signal_info = self.itemData(index)
         return signal_info if isinstance(signal_info, dict) else None
 
     def update_signals_from_signal_classes(self, ndim_filter: int | list[int] | None = None):
@@ -528,6 +537,11 @@ class SignalComboBox(BECWidget, QComboBox):
 
         combo_items: list[str | tuple[str, dict]] = []
         item_tooltips: dict[int, str] = {}
+        for entry_name, entry_config in self._extra_class_signals:
+            if self._store_signal_config:
+                combo_items.append((entry_name, dict(entry_config)))
+            else:
+                combo_items.append(entry_name)
         for device_name, signal_name, signal_config in signals:
             if self._device and device_name != self._device:
                 continue

bec_widgets/widgets/plots/image/toolbar_components/device_selection.py

@@ -6,25 +6,57 @@ from bec_widgets.utils.toolbars.connections import BundleConnection
 from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
 from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox
 
+#: Signal classes supported by the DataAPI-backed plot widgets.
+DEFAULT_SIGNAL_CLASSES = ["PreviewSignal", "AsyncSignal", "AsyncMultiSignal", "DynamicSignal"]
+
+#: Reserved entry name of the scan-less device_monitor_1d stream (DataAPI device plugin).
+MONITOR_1D_ENTRY = "monitor_1d"
+
+#: Signal-config shape offered for the monitor_1d sentinel entry. The capability is not
+#: introspectable from device info, so the entry is offered for every selected device.
+MONITOR_1D_SIGNAL_CONFIG = {
+    "component_name": MONITOR_1D_ENTRY,
+    "obj_name": MONITOR_1D_ENTRY,
+    "signal_class": "DeviceMonitor1D",
+    "describe": {"signal_info": {"ndim": 1}},
+}
+
 
 class DeviceSelection(QWidget):
-    """Device and signal selection widget for image toolbar."""
+    """Device and signal selection widget for plot widget toolbars.
 
-    def __init__(self, parent=None, client=None):
+    Args:
+        parent: Optional parent widget.
+        client: The BEC client instance.
+        signal_classes: Signal class names offered in the comboboxes. Defaults to
+            ``DEFAULT_SIGNAL_CLASSES``.
+        ndim_filter: Dimensionality filter for the signal combobox. Defaults to ``[1, 2]``.
+        include_monitor_1d: If True, offer the reserved ``"monitor_1d"`` stream entry in the
+            signal combobox and additionally list async-readout devices in the device
+            combobox (potential ``device_monitor_1d`` publishers).
+    """
+
+    def __init__(
+        self,
+        parent=None,
+        client=None,
+        signal_classes: list[str] | None = None,
+        ndim_filter: list[int] | None = None,
+        include_monitor_1d: bool = False,
+    ):
         super().__init__(parent=parent)
 
         self.client = client
-        self.supported_signals = [
-            "PreviewSignal",
-            "AsyncSignal",
-            "AsyncMultiSignal",
-            "DynamicSignal",
-        ]
+        self.supported_signals = list(signal_classes or DEFAULT_SIGNAL_CLASSES)
+        ndim_filter = list(ndim_filter or [1, 2])
 
         # Create device combobox with signal class filter
         # This will only show devices that have signals matching the supported signal classes
         self.device_combo_box = DeviceComboBox(
-            parent=self, client=self.client, signal_class_filter=self.supported_signals
+            parent=self,
+            client=self.client,
+            signal_class_filter=self.supported_signals,
+            include_async_readout_devices=include_monitor_1d,
         )
         self.device_combo_box.setToolTip("Select Device")
         self.device_combo_box.setEditable(True)
@@ -32,20 +64,18 @@ class DeviceSelection(QWidget):
         self.device_combo_box.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
         self.device_combo_box.lineEdit().setPlaceholderText("Select Device")
 
-        # Configure SignalComboBox to filter by PreviewSignal and supported async signals
-        # Also filter by ndim (1D and 2D only) for Image widget
+        # Configure SignalComboBox to filter by the supported signal classes and the
+        # requested dimensionality; optionally offer the reserved monitor_1d stream entry.
         self.signal_combo_box = SignalComboBox(
             parent=self,
             client=self.client,
-            signal_class_filter=[
-                "PreviewSignal",
-                "AsyncSignal",
-                "AsyncMultiSignal",
-                "DynamicSignal",
-            ],
-            ndim_filter=[1, 2],  # Only show 1D and 2D signals for Image widget
+            signal_class_filter=self.supported_signals,
+            ndim_filter=ndim_filter,
             store_signal_config=True,
             require_device=True,
+            extra_class_signals=(
+                [(MONITOR_1D_ENTRY, MONITOR_1D_SIGNAL_CONFIG)] if include_monitor_1d else None
+            ),
         )
         self.signal_combo_box.setToolTip("Select Signal")
         self.signal_combo_box.setEditable(True)
@@ -90,7 +120,7 @@ class DeviceSelection(QWidget):
 
                 # Sync signal combobox selection
                 if signal:
-                    # Try to find the signal by component_name (which is what's displayed)
+                    # Try to find the signal by display text, component_name or obj_name
                     found = False
                     for i in range(self.signal_combo_box.count()):
                         text = self.signal_combo_box.itemText(i)
@@ -99,7 +129,8 @@ class DeviceSelection(QWidget):
                         # Check if this matches our signal
                         if config_data:
                             component_name = config_data.get("component_name", "")
-                            if text == component_name or text == signal:
+                            obj_name = config_data.get("obj_name", "")
+                            if signal in (text, component_name, obj_name):
                                 self.signal_combo_box.setCurrentIndex(i)
                                 found = True
                                 break
@@ -144,9 +175,11 @@ class DeviceSelection(QWidget):
         self.signal_combo_box.deleteLater()
 
 
-def device_selection_bundle(components: ToolbarComponents, client=None) -> ToolbarBundle:
+def device_selection_bundle(
+    components: ToolbarComponents, client=None, **selection_kwargs
+) -> ToolbarBundle:
     """
-    Creates a device selection toolbar bundle for Image widget.
+    Creates a device selection toolbar bundle for plot widgets.
 
     Includes a resizable splitter after the device selection. All subsequent bundles'
     actions will appear compactly after the splitter with no gaps.
@@ -154,11 +187,15 @@ def device_selection_bundle(components: ToolbarComponents, client=None) -> Toolb
     Args:
         components (ToolbarComponents): The components to be added to the bundle.
         client: The BEC client instance.
+        **selection_kwargs: Additional keyword arguments passed to ``DeviceSelection``
+            (e.g. ``ndim_filter``, ``include_monitor_1d``).
 
     Returns:
         ToolbarBundle: The device selection toolbar bundle.
     """
-    device_selection_widget = DeviceSelection(parent=components.toolbar, client=client)
+    device_selection_widget = DeviceSelection(
+        parent=components.toolbar, client=client, **selection_kwargs
+    )
     components.add_safe(
         "device_selection", WidgetAction(widget=device_selection_widget, adjust_size=False)
     )

bec_widgets/widgets/plots/multi_waveform/multi_waveform.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
 from collections import deque
-from typing import TYPE_CHECKING, cast
+from typing import Literal
 
 import pyqtgraph as pg
 from bec_lib.logger import bec_logger
 from pydantic import Field, ValidationError, field_validator
-from qtpy.QtCore import Signal
+from qtpy.QtCore import QTimer, Signal
 from qtpy.QtWidgets import QWidget
 
 from bec_widgets.utils.bec_connector import ConnectionConfig
@@ -14,13 +14,16 @@ from bec_widgets.utils.colors import Colors
 from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
 from bec_widgets.utils.qt_data_subscription import QtDataSubscription
 from bec_widgets.utils.side_panel import SidePanel
-from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
+from bec_widgets.utils.toolbars.actions import WidgetAction
+from bec_widgets.widgets.plots.image.toolbar_components.device_selection import (
+    MONITOR_1D_ENTRY,
+    DeviceSelection,
+    DeviceSelectionConnection,
+    device_selection_bundle,
+)
 from bec_widgets.widgets.plots.multi_waveform.settings.control_panel import (
     MultiWaveformControlPanel,
 )
-from bec_widgets.widgets.plots.multi_waveform.toolbar_components.monitor_selection import (
-    monitor_selection_bundle,
-)
 from bec_widgets.widgets.plots.plot_base import PlotBase
 from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
 
@@ -37,7 +40,18 @@ class MultiWaveformConfig(ConnectionConfig):
     flush_buffer: bool | None = Field(
         False, description="Flush the buffer of the plot widget when the curve limit is reached."
     )
-    monitor: str | None = Field(None, description="The monitor to set for the plot widget.")
+    monitor: str | None = Field(None, description="The device to monitor for the plot widget.")
+    monitor_signal: str | None = Field(
+        None,
+        description=(
+            "The signal/entry to monitor on the device. The reserved entry "
+            f"'{MONITOR_1D_ENTRY}' selects the scan-less device_monitor_1d stream."
+        ),
+    )
+    connection_status: Literal["connected", "disconnected", "error"] = Field(
+        "disconnected", description="Current connection status."
+    )
+    connection_error: str | None = Field(None, description="Last connection error, if any.")
     curve_width: int | None = Field(1, description="The width of the curve on the plot.")
     opacity: int | None = Field(50, description="The opacity of the curve on the plot.")
     highlight_last_curve: bool | None = Field(
@@ -73,12 +87,17 @@ class MultiWaveform(PlotBase):
         "max_trace.setter",
         "monitor",
         "monitor.setter",
+        "monitor_signal",
+        "monitor_signal.setter",
         "set_curve_limit",
         "plot",
         "set_curve_highlight",
         "clear_curves",
     ]
 
+    #: Async signal classes routed through the scan-scoped (live-follow) DataAPI path.
+    SUPPORTED_SIGNALS = ["AsyncSignal", "AsyncMultiSignal", "DynamicSignal"]
+
     monitor_signal_updated = Signal()
     highlighted_curve_index_changed = Signal(int)
 
@@ -106,33 +125,53 @@ class MultiWaveform(PlotBase):
         self.visible_curves = []
         self.number_of_visible_curves = 0
 
-        # Data delivery through the DataAPI device stream (monitor_1d).
+        # Data delivery through the DataAPI: scan-less device streams
+        # (monitor_1d, preview signals) or scan-scoped async signals.
         self._data_bridge: QtDataSubscription | None = None
+        self._source_key: tuple[str, str] | None = None
         self._last_ordinal: int | None = None
+        self._signal_config: dict | None = None
+        self._device_selection_updating = False
 
         self._init_multiwaveform_toolbar()
 
+    @property
+    def _config(self) -> MultiWaveformConfig:
+        """Connection-state view consumed by the shared DeviceSelectionConnection."""
+        return self.config
+
     ################################################################################
     # Widget Specific GUI interactions
     ################################################################################
     def _init_multiwaveform_toolbar(self):
-        self.toolbar.add_bundle(
-            monitor_selection_bundle(self.toolbar.components, target_widget=self)
+        bundle = device_selection_bundle(
+            self.toolbar.components, client=self.client, ndim_filter=[1], include_monitor_1d=True
+        )
+        self.toolbar.components.add_safe(
+            "color_map",
+            WidgetAction(
+                widget=BECColorMapWidget(cmap=self.config.color_palette), adjust_size=False
+            ),
+        )
+        bundle.add_action("color_map")
+        self.toolbar.add_bundle(bundle)
+        self.toolbar.connect_bundle(
+            "device_selection",
+            DeviceSelectionConnection(self.toolbar.components, target_widget=self),
         )
         self.toolbar.toggle_action_visibility("reset_legend", visible=False)
 
-        combobox = self.toolbar.components.get_action("monitor_selection").widget
-        combobox.currentTextChanged.connect(self.connect_monitor)
-
         cmap = self.toolbar.components.get_action("color_map").widget
         cmap.colormap_changed_signal.connect(self.change_colormap)
 
         bundles = self.toolbar.shown_bundles
-        bundles.insert(0, "monitor_selection")
+        bundles.insert(0, "device_selection")
         self.toolbar.show_bundles(bundles)
 
         self._init_control_panel()
 
+        QTimer.singleShot(0, self._sync_device_selection)
+
     def _init_control_panel(self):
         control_panel = SidePanel(self, orientation="top", panel_max_width=90)
         self.layout_manager.add_widget_relative(control_panel, self.round_plot_widget, "bottom")
@@ -147,13 +186,67 @@ class MultiWaveform(PlotBase):
         control_panel.toolbar.components.get_action("control").action.trigger()
 
     @SafeSlot()
-    def connect_monitor(self, _):
-        combobox = self.toolbar.components.get_action("monitor_selection").widget
-        monitor = combobox.currentText()
+    def on_device_selection_changed(self, _):
+        """
+        Called when device or signal selection changes in the toolbar.
+        This reads from the toolbar and updates the widget properties.
+        """
+        if self._device_selection_updating:
+            return
 
-        if monitor != "":
-            if monitor != self.config.monitor:
-                self.config.monitor = monitor
+        self._device_selection_updating = True
+        try:
+            try:
+                action = self.toolbar.components.get_action("device_selection")
+            except Exception:
+                return
+
+            if action is None:
+                return
+
+            device_selection: DeviceSelection = action.widget
+            device = device_selection.device_combo_box.currentText()
+            signal_text = device_selection.signal_combo_box.currentText()
+
+            if not device:
+                self.device = ""
+                return
+            if not device_selection.device_combo_box.is_valid_input:
+                return
+
+            if not device_selection.signal_combo_box.is_valid_input:
+                if self.config.monitor_signal:
+                    self.signal = ""
+                if device != (self.config.monitor or ""):
+                    self.device = device
+                return
+
+            if device == self.config.monitor and signal_text == self.config.monitor_signal:
+                return
+
+            # Get the signal config stored in the combobox
+            signal_config = device_selection.signal_combo_box.get_signal_config()
+
+            if not signal_config and signal_text != MONITOR_1D_ENTRY:
+                # Fallback: try to get config from device
+                try:
+                    device_obj = self.dev[device]
+                    signal_config = device_obj._info["signals"].get(signal_text, {})
+                except (KeyError, AttributeError):
+                    logger.warning(f"Could not get signal config for {device}.{signal_text}")
+                    signal_config = None
+
+            # Store signal config and set properties which will trigger the connection
+            pending_config = signal_config if isinstance(signal_config, dict) else None
+            if device != (self.config.monitor or ""):
+                self._signal_config = pending_config
+                self.device = device
+            # The device setter may replace the pending config while reconnecting
+            # the previous signal; restore it for the signal connection.
+            self._signal_config = pending_config
+            self.signal = signal_text
+        finally:
+            self._device_selection_updating = False
 
     @SafeSlot(str)
     def change_colormap(self, colormap: str):
@@ -221,7 +314,7 @@ class MultiWaveform(PlotBase):
         except ValidationError:
             return
         self.set_curve_highlight(self._current_highlight_index)
-        self._sync_monitor_selection_toolbar()
+        self._sync_colormap_toolbar()
 
     @SafeProperty(int)
     def opacity(self) -> int:
@@ -278,41 +371,195 @@ class MultiWaveform(PlotBase):
             max_trace=self.config.curve_limit, flush_buffer=self.config.flush_buffer
         )
 
+    ################################################################################
+    # Data Acquisition
+    ################################################################################
+
+    @SafeProperty(str, auto_emit=True, designable=False)
+    def device(self) -> str:
+        """
+        The name of the device to monitor for waveform data.
+        """
+        return self.config.monitor or ""
+
+    @device.setter
+    def device(self, value: str):
+        """
+        Set the device name for the plot. This should be used together with signal.
+        When both device and signal are set, the widget connects to that device signal.
+
+        Args:
+            value(str): The name of the device to monitor.
+        """
+        if not value:
+            # Clear the monitor if empty device name
+            if self.config.monitor:
+                self._cleanup_data_api_subscription()
+                self.config.monitor = None
+                self.config.monitor_signal = None
+                self._signal_config = None
+                self._set_connection_status("disconnected")
+            return
+
+        old_device = self.config.monitor
+        if old_device and self.config.monitor_signal and old_device != value:
+            self._cleanup_data_api_subscription()
+        self.config.monitor = value
+
+        # If we have a signal, reconnect with the new device
+        signal = self.config.monitor_signal
+        if not signal:
+            return
+        if signal == MONITOR_1D_ENTRY:
+            self._signal_config = None
+            self._setup_data_api_subscription()
+            return
+        try:
+            device_obj = self.dev[value]
+            signal_config = device_obj._info.get("signals", {}).get(signal)
+        except (KeyError, AttributeError):
+            logger.warning(f"Device '{value}' not found")
+            self._cleanup_data_api_subscription()
+            self._set_connection_status("error", f"Device '{value}' not found")
+            return
+        if isinstance(signal_config, dict) and signal_config.get("signal_class"):
+            self._signal_config = signal_config
+            self._setup_data_api_subscription()
+        else:
+            logger.warning(f"Signal '{signal}' doesn't exist on device '{value}'")
+            self._cleanup_data_api_subscription()
+            self.config.monitor_signal = None
+            self._signal_config = None
+            self._set_connection_status("error", f"Signal '{signal}' doesn't exist")
+
+    @SafeProperty(str, auto_emit=True, designable=False)
+    def signal(self) -> str:
+        """
+        The signal/entry name to monitor on the device.
+        """
+        return self.config.monitor_signal or ""
+
+    @signal.setter
+    def signal(self, value: str):
+        """
+        Set the device signal for the plot. This should be used together with device.
+        When set, it will connect to updates from that device signal.
+
+        Args:
+            value(str): The signal name to monitor. The reserved entry ``"monitor_1d"``
+                selects the scan-less device_monitor_1d stream.
+        """
+        if not value:
+            if self.config.monitor_signal:
+                self._cleanup_data_api_subscription()
+                self.config.monitor_signal = None
+                self._signal_config = None
+                self._set_connection_status("disconnected")
+            return
+
+        old_signal = self.config.monitor_signal
+        if self.config.monitor and old_signal and old_signal != value:
+            self._cleanup_data_api_subscription()
+        self.config.monitor_signal = value
+
+        if not self.config.monitor:
+            logger.debug(f"signal setter: No device set yet for signal '{value}'")
+            return
+
+        if value == MONITOR_1D_ENTRY:
+            self._signal_config = None
+            self._setup_data_api_subscription()
+            return
+
+        try:
+            device_obj = self.dev[self.config.monitor]
+            signal_config = device_obj._info["signals"].get(value)
+        except (KeyError, AttributeError):
+            signal_config = None
+        if not isinstance(signal_config, dict) or not signal_config.get("signal_class"):
+            # Fall back to a combobox-provided config for this entry, if any
+            signal_config = self._signal_config
+            if not self._signal_config_matches(signal_config, value):
+                logger.warning(
+                    f"Could not find valid configuration for signal '{value}' "
+                    f"on device '{self.config.monitor}'."
+                )
+                self._signal_config = None
+                self._set_connection_status("error", f"Signal '{value}' not found")
+                return
+        self._signal_config = signal_config
+        self._setup_data_api_subscription()
+
     @SafeProperty(str)
     def monitor(self) -> str:
         """
-        The monitor of the figure widget.
+        The monitored device of the figure widget (alias of ``device``).
         """
         return self.config.monitor
 
     @monitor.setter
     def monitor(self, value: str):
         """
-        Set the monitor of the figure widget.
+        Set the monitored device and connect to it, auto-selecting the signal.
 
         Args:
-            value(str): The monitor to set.
+            value(str): The device to monitor.
         """
         self.plot(value)
 
+    @SafeProperty(str)
+    def monitor_signal(self) -> str:
+        """
+        The monitored signal/entry on the device (alias of ``signal``).
+        """
+        return self.config.monitor_signal
+
+    @monitor_signal.setter
+    def monitor_signal(self, value: str):
+        """
+        Set the monitored signal/entry on the device.
+
+        Args:
+            value(str): The signal to monitor.
+        """
+        self.signal = value
+
     ################################################################################
     # High Level methods for API
     ################################################################################
 
     @SafeSlot(popup_error=True)
-    def plot(self, monitor: str, color_palette: str | None = "plasma"):
+    def plot(self, monitor: str, signal: str | None = None, color_palette: str | None = "plasma"):
         """
-        Create a plot for the given monitor.
+        Create a plot for the given monitor device and signal.
+
         Args:
-            monitor (str): The monitor to set.
+            monitor (str): The device to monitor.
+            signal (str|None): The signal/entry to monitor on the device. The reserved
+                entry ``"monitor_1d"`` selects the scan-less device_monitor_1d stream.
+                If None, the signal is auto-selected: the device's only 1D-capable
+                preview/async signal if unambiguous, the monitor_1d stream otherwise.
             color_palette (str|None): The color palette to use for the plot.
         """
+        if not monitor:
+            self.device = ""
+            return
         self.entry_validator.validate_monitor(monitor)
-        self.config.monitor = monitor
-        self._setup_data_api_subscription()
+        if signal is None:
+            signal = self._default_signal_for(monitor)
+        if self.config.monitor_signal and self.config.monitor != monitor:
+            # Clear the old entry first to avoid reconnect attempts on the new device
+            self.signal = ""
+        # Guard the device update so the toolbar sync does not pull the
+        # auto-selected first combobox entry before the requested signal is set.
+        self._device_selection_updating = True
+        try:
+            self.device = monitor
+        finally:
+            self._device_selection_updating = False
+        self.signal = signal
         if color_palette is not None:
             self.color_palette = color_palette
-        self._sync_monitor_selection_toolbar()
 
     @SafeSlot(int, bool)
     def set_curve_limit(self, max_trace: int, flush_buffer: bool):
@@ -354,31 +601,49 @@ class MultiWaveform(PlotBase):
     @SafeSlot(object)
     def _on_data_update(self, update) -> None:
         """
-        Render one columnar DataAPI update of the monitor_1d device stream.
+        Render one columnar DataAPI update of the monitored source.
 
         Each value of the source is one 1-D trace (newest last); ordinals are
-        arrival counters, so only traces newer than the last rendered ordinal
-        are appended to the curve deque.
+        arrival counters (device streams) or per-scan async ordinals, so only
+        traces newer than the last rendered ordinal are appended to the curve
+        deque. ``"replace"`` async sources expose a single current state which
+        replaces the trace set.
 
         Args:
-            update (SubscriptionUpdate): Full-state snapshot of the monitor
-                stream (standalone group).
+            update (SubscriptionUpdate): Full-state snapshot of the monitored
+                source.
         """
-        monitor = self.config.monitor
-        if not monitor:
+        if self._source_key is None:
             return
-        source = update.get(monitor, "monitor_1d")
+        source = update.get(*self._source_key)
         if source is None or not source.values:
             return
 
-        current_scan_id = source.metadata.get("scan_id", None)
+        current_scan_id = self._effective_scan_id(update, source)
         if current_scan_id != self.scan_id:
+            self.old_scan_id = self.scan_id
             self.scan_id = current_scan_id
             self.clear_curves()
             self.curves.clear()
+            if update.scan_id:
+                # Scan-scoped async ordinals restart with every scan; the
+                # arrival counters of scan-less device streams do not.
+                self._last_ordinal = None
             if self.crosshair:
                 self.crosshair.clear_markers()
 
+        if source.metadata.get("async_update_type") == "replace":
+            # A replace source exposes one point: its current full state.
+            self.clear_curves()
+            self.curves.clear()
+            curve = pg.PlotDataItem()
+            curve.setData(source.values[-1])
+            self.plot_item.addItem(curve)
+            self.curves.append(curve)
+            self._last_ordinal = None
+            self.set_curve_limit(self.config.curve_limit, self.config.flush_buffer)
+            return
+
         last_ordinal = self._last_ordinal
         new_traces = [
             (ordinal, data)
@@ -398,6 +663,24 @@ class MultiWaveform(PlotBase):
         # Max Trace and scale colors
         self.set_curve_limit(self.config.curve_limit, self.config.flush_buffer)
 
+    @staticmethod
+    def _effective_scan_id(update, source) -> str | None:
+        """
+        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 device streams.
+
+        Args:
+            update (SubscriptionUpdate): The update snapshot.
+            source (SourceData): The rendered source of the update.
+
+        Returns:
+            str | None: The scan id, or None if not known (yet).
+        """
+        if update.scan_id:
+            return update.scan_id
+        return source.metadata.get("scan_id")
+
     @SafeSlot(int)
     def set_curve_highlight(self, index: int):
         """
@@ -435,29 +718,143 @@ class MultiWaveform(PlotBase):
 
         self.highlighted_curve_index_changed.emit(self._current_highlight_index)
 
+    ################################################################################
+    # Signal classification and DataAPI routing
+    ################################################################################
+
+    def _default_signal_for(self, device: str) -> str:
+        """
+        Auto-select a signal for a device-only selection.
+
+        Returns the device's only 1D-capable preview/async signal if it is
+        unambiguous; the reserved monitor_1d stream entry otherwise (legacy
+        monitor devices, or devices with several 1D-capable signals).
+
+        Args:
+            device (str): The device name.
+
+        Returns:
+            str: The signal/entry to monitor.
+        """
+        try:
+            device_obj = self.dev[device]
+            signals = device_obj._info.get("signals", {})
+        except (KeyError, AttributeError):
+            signals = {}
+        supported = {"PreviewSignal", *self.SUPPORTED_SIGNALS}
+        candidates = []
+        for name, signal_config in signals.items():
+            if not isinstance(signal_config, dict):
+                continue
+            if signal_config.get("signal_class") not in supported:
+                continue
+            describe = signal_config.get("describe") or {}
+            signal_info = describe.get("signal_info") or {}
+            if signal_info.get("ndim") == 1:
+                candidates.append(name)
+        if len(candidates) == 1:
+            return candidates[0]
+        if candidates:
+            logger.warning(
+                f"Device '{device}' has multiple 1D-capable signals ({candidates}); "
+                f"defaulting to the '{MONITOR_1D_ENTRY}' stream. Select a signal "
+                "explicitly to plot one of them."
+            )
+        return MONITOR_1D_ENTRY
+
+    @staticmethod
+    def _signal_config_matches(signal_config: dict | None, signal: str) -> bool:
+        """Whether a stored signal config describes the given signal/entry."""
+        if not isinstance(signal_config, dict) or not signal_config.get("signal_class"):
+            return False
+        return signal in (signal_config.get("component_name"), signal_config.get("obj_name"))
+
+    def _resolve_route(self, device: str, signal: str) -> tuple[str | None, str, int | None] | None:
+        """
+        Classify the selected (device, signal) into a DataAPI route.
+
+        Routing table:
+            monitor_1d sentinel -> scan=None, entry "monitor_1d" (device stream)
+            PreviewSignal (1D)  -> scan=None, entry = signal (device stream)
+            Async*/DynamicSignal (1D) -> scan="live", entry = obj_name (live-follow)
+
+        Args:
+            device (str): The device name.
+            signal (str): The selected signal/entry.
+
+        Returns:
+            tuple | None: ``(scan, entry, max_points)`` or None if the selection
+                cannot be served (connection status is set accordingly).
+        """
+        if signal == MONITOR_1D_ENTRY:
+            return (None, MONITOR_1D_ENTRY, self.config.curve_limit)
+
+        signal_config = self._signal_config or {}
+        signal_class = signal_config.get("signal_class")
+        supported_classes = ["PreviewSignal"] + self.SUPPORTED_SIGNALS
+        if signal_class not in supported_classes:
+            logger.warning(
+                f"Signal '{device}.{signal}' has unsupported signal class '{signal_class}'. "
+                f"Supported classes: {supported_classes}"
+            )
+            self._set_connection_status("error", f"Unsupported signal class '{signal_class}'")
+            return None
+
+        describe = signal_config.get("describe") or {}
+        signal_info = describe.get("signal_info") or {}
+        ndim = signal_info.get("ndim", None)
+        if ndim != 1:
+            logger.warning(f"Unsupported ndim '{ndim}' for monitor '{device}.{signal}'.")
+            self._set_connection_status("error", f"Unsupported ndim '{ndim}'")
+            return None
+
+        if signal_class == "PreviewSignal":
+            # Scan-less device stream served by the DataAPI device plugin.
+            return (None, signal, self.config.curve_limit)
+
+        # Scan-scoped async stream; the DataAPI rebinds on new scans and hands
+        # terminal scans over to history automatically (live-follow).
+        entry = signal_config.get("obj_name") or f"{device}_{signal}"
+        return ("live", entry, None)
+
     def _setup_data_api_subscription(self):
-        """(Re)create the scan-less DataAPI subscription for the configured monitor."""
+        """(Re)create the DataAPI subscription for the configured device and signal."""
         self._cleanup_data_api_subscription()
-        monitor = self.config.monitor
-        if not monitor:
+        device = self.config.monitor
+        signal = self.config.monitor_signal
+        if not device or not signal:
+            self._set_connection_status("disconnected")
             return
+        route = self._resolve_route(device, signal)
+        if route is None:
+            return
+        scan, entry, max_points = route
         try:
             self._data_bridge = QtDataSubscription(
                 self.client,
-                sources=[(monitor, "monitor_1d")],
-                scan=None,
+                sources=[(device, entry)],
+                scan=scan,
                 parent=self,
                 min_emit_interval=0.1,
-                max_points=self.config.curve_limit,
+                max_points=max_points,
             )
             self._data_bridge.updated.connect(self._on_data_update)
-            self.connected = True
-        except Exception as exc:
-            logger.warning(f"Failed to configure multi waveform data subscription: {exc}")
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(
+                f"Failed to configure multi waveform data subscription for "
+                f"{device}.{signal}: {exc}"
+            )
             self._cleanup_data_api_subscription()
+            self._set_connection_status("error", str(exc))
+            return
+        self._source_key = (device, entry)
+        self.connected = True
+        self._set_connection_status("connected")
+        logger.info(f"Connected to {device}.{signal} via {'live' if scan else 'device'} scope")
 
     def _cleanup_data_api_subscription(self):
         self._last_ordinal = None
+        self._source_key = None
         self.connected = False
         if self._data_bridge is None:
             return
@@ -466,6 +863,12 @@ class MultiWaveform(PlotBase):
         finally:
             self._data_bridge = None
 
+    def _set_connection_status(self, status: str, message: str | None = None) -> None:
+        self.config.connection_status = status
+        self.config.connection_error = message
+        self.property_changed.emit("connection_status", status)
+        self.property_changed.emit("connection_error", message or "")
+
     ################################################################################
     # Utility Methods
     ################################################################################
@@ -500,29 +903,81 @@ class MultiWaveform(PlotBase):
         for item in items_to_remove:
             self.plot_item.removeItem(item)
 
-    def _sync_monitor_selection_toolbar(self):
+    def _sync_device_selection(self):
         """
-        Sync the motor map selection toolbar with the current motor map.
+        Synchronize the device and signal comboboxes with the current monitor state.
+        This ensures the toolbar reflects the device and signal properties.
         """
+        try:
+            device_selection_action = self.toolbar.components.get_action("device_selection")
+        except Exception:  # noqa: BLE001 - toolbar might not be ready during early init
+            logger.warning(f"MultiWaveform ({self.object_name}) toolbar was not ready during init.")
+            return
 
-        combobox_widget: DeviceComboBox = cast(
-            DeviceComboBox, self.toolbar.components.get_action("monitor_selection").widget
-        )
-        cmap_widget: BECColorMapWidget = cast(
-            BECColorMapWidget, self.toolbar.components.get_action("color_map").widget
-        )
+        if device_selection_action is None:
+            return
 
-        monitor = combobox_widget.currentText()
-        color_palette = cmap_widget.colormap
+        device_selection: DeviceSelection = device_selection_action.widget
+        target_device = self.config.monitor or ""
+        target_entry = self.config.monitor_signal or ""
 
-        if monitor != self.config.monitor:
-            combobox_widget.setCurrentText(monitor)
-            combobox_widget.blockSignals(True)
-            combobox_widget.set_device(self.config.monitor)
-            combobox_widget.check_validity(self.config.monitor)
-            combobox_widget.blockSignals(False)
+        # Check if already synced
+        if (
+            device_selection.device_combo_box.currentText() == target_device
+            and device_selection.signal_combo_box.currentText() == target_entry
+        ):
+            return
 
-        if color_palette != self.config.color_palette:
+        device_selection.set_device_and_signal(target_device, target_entry)
+
+    def _sync_signal_from_toolbar(self) -> None:
+        """
+        Pull the signal selection from the toolbar if it differs from the current signal.
+        This keeps CLI-driven device updates in sync with the signal combobox state.
+        """
+        if self._device_selection_updating:
+            return
+
+        if not self.config.monitor:
+            return
+
+        try:
+            device_selection_action = self.toolbar.components.get_action("device_selection")
+        except Exception:  # noqa: BLE001 - toolbar might not be ready during early init
+            return
+
+        if device_selection_action is None:
+            return
+
+        device_selection: DeviceSelection = device_selection_action.widget
+        if device_selection.device_combo_box.currentText() != self.config.monitor:
+            return
+
+        signal_text = device_selection.signal_combo_box.currentText()
+        if not signal_text or signal_text == self.config.monitor_signal:
+            return
+
+        signal_config = device_selection.signal_combo_box.get_signal_config()
+        if signal_config:
+            self._signal_config = signal_config
+        elif signal_text != MONITOR_1D_ENTRY:
+            return
+
+        self._device_selection_updating = True
+        try:
+            self.signal = signal_text
+        finally:
+            self._device_selection_updating = False
+
+    def _sync_colormap_toolbar(self):
+        """
+        Sync the colormap toolbar widget with the current color palette.
+        """
+        try:
+            cmap_widget: BECColorMapWidget = self.toolbar.components.get_action("color_map").widget
+        except Exception:  # noqa: BLE001 - toolbar might not be ready during early init
+            return
+        if cmap_widget.colormap != self.config.color_palette:
             cmap_widget.blockSignals(True)
             cmap_widget.colormap = self.config.color_palette
             cmap_widget.blockSignals(False)
@@ -530,4 +985,8 @@ class MultiWaveform(PlotBase):
     def cleanup(self):
         self._cleanup_data_api_subscription()
         self.clear_curves()
+        try:
+            self.toolbar.disconnect_bundle("device_selection")
+        except Exception:  # noqa: BLE001
+            pass
         super().cleanup()

bec_widgets/widgets/plots/multi_waveform/toolbar_components/monitor_selection.py

@@ -1,100 +0,0 @@
-from bec_lib.device import ReadoutPriority
-from qtpy.QtCore import Qt
-from qtpy.QtWidgets import QStyledItemDelegate, QWidget
-
-from bec_widgets.utils.error_popups import SafeSlot
-from bec_widgets.utils.toolbars.actions import DeviceComboBoxAction, WidgetAction
-from bec_widgets.utils.toolbars.bundles import ToolbarComponents
-from bec_widgets.utils.toolbars.toolbar import ToolbarBundle
-from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
-    BECDeviceFilter,
-    DeviceComboBox,
-)
-from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
-
-
-class NoCheckDelegate(QStyledItemDelegate):
-    """To reduce space in combo boxes by removing the checkmark."""
-
-    def initStyleOption(self, option, index):
-        super().initStyleOption(option, index)
-        # Remove any check indicator
-        option.checkState = Qt.Unchecked
-
-
-def monitor_selection_bundle(
-    components: ToolbarComponents, target_widget: QWidget
-) -> ToolbarBundle:
-    """
-    Creates a monitor selection toolbar bundle.
-
-    Args:
-        components (ToolbarComponents): The components to be added to the bundle.
-
-    Returns:
-        ToolbarBundle: The monitor selection toolbar bundle.
-    """
-    components.add_safe(
-        "monitor_selection",
-        DeviceComboBoxAction(
-            target_widget=target_widget,
-            device_filter=[BECDeviceFilter.DEVICE],
-            readout_priority_filter=ReadoutPriority.ASYNC,
-            add_empty_item=True,
-            no_check_delegate=True,
-        ),
-    )
-    components.add_safe(
-        "color_map", WidgetAction(widget=BECColorMapWidget(cmap="plasma"), adjust_size=False)
-    )
-    bundle = ToolbarBundle("monitor_selection", components)
-    bundle.add_action("monitor_selection")
-    bundle.add_action("color_map")
-    return bundle
-
-
-class MultiWaveformSelectionToolbarBundle(ToolbarBundle):
-    """
-    A bundle of actions for a toolbar that selects motors.
-    """
-
-    def __init__(self, bundle_id="monitor_selection", target_widget=None, **kwargs):
-        super().__init__(bundle_id=bundle_id, actions=[], **kwargs)
-        self.target_widget = target_widget
-
-        # Monitor Selection
-        self.monitor = DeviceComboBox(
-            device_filter=BECDeviceFilter.DEVICE, readout_priority_filter=ReadoutPriority.ASYNC
-        )
-        self.monitor.addItem("", None)
-        self.monitor.setCurrentText("")
-        self.monitor.setToolTip("Select Monitor")
-        self.monitor.setItemDelegate(NoCheckDelegate(self.monitor))
-        self.add_action("monitor", WidgetAction(widget=self.monitor, adjust_size=False))
-
-        # Colormap Selection
-        self.colormap_widget = BECColorMapWidget(cmap="plasma")
-        self.add_action("color_map", WidgetAction(widget=self.colormap_widget, adjust_size=False))
-
-        # Connect slots, a device will be connected upon change of any combobox
-        self.monitor.currentTextChanged.connect(lambda: self.connect())
-        self.colormap_widget.colormap_changed_signal.connect(self.change_colormap)
-
-    @SafeSlot()
-    def connect(self):
-        monitor = self.monitor.currentText()
-
-        if monitor != "":
-            if monitor != self.target_widget.config.monitor:
-                self.target_widget.monitor = monitor
-
-    @SafeSlot(str)
-    def change_colormap(self, colormap: str):
-        self.target_widget.color_palette = colormap
-
-    def cleanup(self):
-        """
-        Cleanup the toolbar bundle.
-        """
-        self.monitor.close()
-        self.monitor.deleteLater()

tests/unit_tests/test_multi_waveform_next_gen.py

@@ -13,6 +13,22 @@ from .conftest import create_widget
 ##################################################
 
 
+def _set_signal_config(
+    client, device: str, signal_name: str, signal_class: str, ndim: int, obj_name: str | None = None
+):
+    device = client.device_manager.devices[device]
+    device._info["signals"][signal_name] = {
+        "obj_name": obj_name or signal_name,
+        "signal_class": signal_class,
+        "component_name": signal_name,
+        "describe": {"signal_info": {"ndim": ndim}},
+    }
+
+
+def _clear_signal_config(client, device: str, signal_name: str):
+    client.device_manager.devices[device]._info["signals"].pop(signal_name, None)
+
+
 def _fake_bridge_factory(monkeypatch):
     created = []
 
@@ -65,6 +81,31 @@ def _monitor_update(traces, scan_id="scan_1", monitor="waveform1d", start=0):
     )
 
 
+def _async_update(
+    traces, scan_id="scan_1", device="eiger", entry="eiger_data", start=0, update_type="add"
+):
+    """Build a full-state scan-scoped async snapshot: one trace per ordinal."""
+    ordinals = tuple(range(start, start + len(traces)))
+    source = SourceData(
+        device=device,
+        entry=entry,
+        kind="async",
+        ordinals=ordinals,
+        values=tuple(traces),
+        timestamps=tuple(float(ordinal) for ordinal in ordinals),
+        complete=True,
+        metadata={"async_update_type": update_type},
+    )
+    return SubscriptionUpdate(
+        scan_id=scan_id,
+        reason="live",
+        sources={(device, entry): source},
+        aligned_ordinals=ordinals,
+        complete=True,
+        metadata={"group": "standalone"},
+    )
+
+
 ##################################################
 # MultiWaveform widget base functionality tests
 ##################################################
@@ -85,6 +126,7 @@ def test_multiwaveform_initialization(qtbot, mocked_client):
     assert len(mw.plot_item.curves) == 0
     # Multiwaveform specific
     assert mw.monitor is None
+    assert mw.monitor_signal is None
     assert mw.color_palette == "plasma"
     assert mw.max_trace == 200
     assert mw.flush_buffer is False
@@ -93,17 +135,22 @@ def test_multiwaveform_initialization(qtbot, mocked_client):
     assert mw.scan_id is None
     assert mw.highlighted_index == 0
     assert mw._data_bridge is None
+    assert mw.config.connection_status == "disconnected"
 
 
 def test_multiwaveform_set_monitor(qtbot, mocked_client):
     mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
     assert mw.monitor is None
 
-    # Set a monitor; data flows through a scan-less DataAPI subscription.
+    # Set a monitor without a signal; the device has no 1D-capable class signals,
+    # so the selection falls back to the scan-less monitor_1d stream.
     mw.plot("waveform1d")
     assert mw.monitor == "waveform1d"
+    assert mw.monitor_signal == "monitor_1d"
     assert mw.config.monitor == "waveform1d"
+    assert mw.config.monitor_signal == "monitor_1d"
     assert mw.connected is True
+    assert mw.config.connection_status == "connected"
     assert mw._data_bridge is not None
     assert mw._data_bridge.sources == [("waveform1d", "monitor_1d")]
     assert mw._data_bridge.scan_id == ""  # device scope
@@ -132,6 +179,172 @@ def test_multiwaveform_bridge_lifecycle(qtbot, mocked_client, monkeypatch):
     assert mw._data_bridge is None
 
 
+##################################################
+# Signal classification and routing
+##################################################
+
+
+def test_multiwaveform_async_signal_routes_live(qtbot, mocked_client, monkeypatch):
+    """Selecting an AsyncSignal creates a scan="live" bridge with the obj_name entry.
+
+    This is the simulated-waveform case: the device publishes via
+    device_async_signal (scan-scoped), not via the device_monitor_1d stream.
+    """
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="eiger_data"
+    )
+
+    mw.plot("eiger", "data")
+    assert mw.monitor == "eiger"
+    assert mw.monitor_signal == "data"
+    assert mw.connected is True
+    assert bridges[-1].scan == "live"
+    assert bridges[-1].sources == [("eiger", "eiger_data")]
+    assert bridges[-1].max_points is None
+    _clear_signal_config(mocked_client, "eiger", "data")
+
+
+def test_multiwaveform_device_only_picks_unambiguous_async_signal(
+    qtbot, mocked_client, monkeypatch
+):
+    """A device-only plot() picks the device's only 1D-capable class signal."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="eiger_data"
+    )
+
+    mw.plot("eiger")
+    assert mw.monitor_signal == "data"
+    assert bridges[-1].scan == "live"
+    assert bridges[-1].sources == [("eiger", "eiger_data")]
+    _clear_signal_config(mocked_client, "eiger", "data")
+
+
+def test_multiwaveform_device_only_ambiguous_falls_back_to_monitor(
+    qtbot, mocked_client, monkeypatch
+):
+    """A device with several 1D-capable signals falls back to the monitor_1d stream."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="eiger_data"
+    )
+    _set_signal_config(
+        mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=1, obj_name="eiger_img"
+    )
+
+    mw.plot("eiger")
+    assert mw.monitor_signal == "monitor_1d"
+    assert bridges[-1].scan is None
+    assert bridges[-1].sources == [("eiger", "monitor_1d")]
+    _clear_signal_config(mocked_client, "eiger", "data")
+    _clear_signal_config(mocked_client, "eiger", "img")
+
+
+def test_multiwaveform_preview_signal_routes_scanless(qtbot, mocked_client, monkeypatch):
+    """Selecting a PreviewSignal creates a scan-less bridge with the signal entry."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client,
+        "waveform1d",
+        "wave_prev",
+        signal_class="PreviewSignal",
+        ndim=1,
+        obj_name="waveform1d_wave_prev",
+    )
+
+    mw.plot("waveform1d", "wave_prev")
+    assert mw.monitor_signal == "wave_prev"
+    assert bridges[-1].scan is None
+    assert bridges[-1].sources == [("waveform1d", "wave_prev")]
+    assert bridges[-1].max_points == mw.config.curve_limit
+    _clear_signal_config(mocked_client, "waveform1d", "wave_prev")
+
+
+def test_multiwaveform_rejects_unsupported_signal(qtbot, mocked_client, monkeypatch):
+    """A signal with an unsupported class does not create a bridge and flags an error."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(mocked_client, "eiger", "plain", signal_class="Signal", ndim=1)
+
+    mw.plot("eiger", "plain")
+    assert bridges == []
+    assert mw.connected is False
+    assert mw.config.connection_status == "error"
+    _clear_signal_config(mocked_client, "eiger", "plain")
+
+
+def test_multiwaveform_rejects_2d_signal(qtbot, mocked_client, monkeypatch):
+    """A 2D async signal is rejected for the 1D multi waveform plot."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(mocked_client, "eiger", "img2d", signal_class="AsyncSignal", ndim=2)
+
+    mw.plot("eiger", "img2d")
+    assert bridges == []
+    assert mw.config.connection_status == "error"
+    _clear_signal_config(mocked_client, "eiger", "img2d")
+
+
+def test_multiwaveform_switching_device_replaces_async_bridge(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="async_obj"
+    )
+    _set_signal_config(
+        mocked_client,
+        "async_device",
+        "data",
+        signal_class="AsyncSignal",
+        ndim=1,
+        obj_name="async_obj",
+    )
+
+    def fake_get(signal_class_filter):
+        # Production-like discovery so the signal combobox lists the async entry
+        # for both devices when the toolbar re-populates on device change.
+        return [
+            (device, "data", mocked_client.device_manager.devices[device]._info["signals"]["data"])
+            for device in ("eiger", "async_device")
+        ]
+
+    monkeypatch.setattr(mocked_client.device_manager, "get_bec_signals", fake_get)
+
+    mw.plot("eiger", "data")
+    first = bridges[-1]
+    assert first.sources == [("eiger", "async_obj")]
+
+    mw.device = "async_device"
+    assert first.closed is True
+    assert bridges[-1].sources == [("async_device", "async_obj")]
+    _clear_signal_config(mocked_client, "eiger", "data")
+    _clear_signal_config(mocked_client, "async_device", "data")
+
+
+def test_multiwaveform_disconnect_via_empty_device(qtbot, mocked_client, monkeypatch):
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
+    assert mw.connected is True
+
+    mw.device = ""
+    assert bridges[-1].closed is True
+    assert mw.connected is False
+    assert mw.monitor is None
+    assert mw.monitor_signal is None
+    assert mw.config.connection_status == "disconnected"
+
+
+##################################################
+# Rendering
+##################################################
+
+
 def test_multiwaveform_set_properties(qtbot, mocked_client):
     """Check that MultiWaveform properties can be set and retrieved correctly."""
     mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
@@ -253,6 +466,58 @@ def test_multiwaveform_scan_change_clears_curves(qtbot, mocked_client, monkeypat
     assert np.array_equal(y_data, new_trace)
 
 
+def test_multiwaveform_async_scan_rollover_clears_curves(qtbot, mocked_client, monkeypatch):
+    """For scan-scoped async sources the update.scan_id drives the rollover and
+    resets the ordinal delta-append (async ordinals restart per scan)."""
+    _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="eiger_data"
+    )
+    mw.plot("eiger", "data")
+
+    first = np.array([1, 2, 3])
+    second = np.array([4, 5, 6])
+    mw._on_data_update(_async_update([first, second], scan_id="scan_1"))
+    assert mw.scan_id == "scan_1"
+    assert len(mw.curves) == 2
+    assert mw._last_ordinal == 1
+
+    # New scan: ordinals restart at 0; curves are cleared and the delta-append reset.
+    rollover = np.array([7, 8, 9])
+    mw._on_data_update(_async_update([rollover], scan_id="scan_2"))
+    assert mw.scan_id == "scan_2"
+    assert len(mw.curves) == 1
+    assert mw._last_ordinal == 0
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, rollover)
+    _clear_signal_config(mocked_client, "eiger", "data")
+
+
+def test_multiwaveform_async_replace_updates_single_trace(qtbot, mocked_client, monkeypatch):
+    """A "replace" async source exposes one current state that replaces the trace set."""
+    _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    _set_signal_config(
+        mocked_client, "eiger", "data", signal_class="AsyncSignal", ndim=1, obj_name="eiger_data"
+    )
+    mw.plot("eiger", "data")
+
+    first = np.array([1, 2, 3])
+    second = np.array([4, 5, 6])
+    mw._on_data_update(_async_update([first], scan_id="scan_1", update_type="replace"))
+    assert len(mw.curves) == 1
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, first)
+
+    # The replace source keeps ordinal 0; the new state still replaces the trace.
+    mw._on_data_update(_async_update([second], scan_id="scan_1", update_type="replace"))
+    assert len(mw.curves) == 1
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, second)
+    _clear_signal_config(mocked_client, "eiger", "data")
+
+
 def test_multiwaveform_highlight_last_curve(qtbot, mocked_client, monkeypatch):
     """Check highlight_last_curve behavior."""
     _fake_bridge_factory(monkeypatch)
@@ -309,7 +574,7 @@ def test_multiwaveform_set_colormap(qtbot, mocked_client, monkeypatch):
         traces.append(np.array([i, i + 1, i + 2]))
         mw._on_data_update(_monitor_update(list(traces)))
 
-    # Default color_palette is "magma"
+    # Default color_palette is "plasma"
     assert mw.color_palette == "plasma"
     # Now change to a new colormap
     mw.color_palette = "viridis"
@@ -388,23 +653,108 @@ def test_widget_updates_control_panel(qtbot, mocked_client):
     assert mw.controls.ui.highlight_last_curve.isChecked() is False
 
 
-def test_selection_toolbar_updates_widget(qtbot, mocked_client):
-    """
-    Confirm that selecting a monitor and a colormap from the selection toolbar
-    updates the widget properties.
-    """
+def test_selection_toolbar_builds_device_and_signal_comboboxes(qtbot, mocked_client):
+    """The toolbar carries the shared device+signal selection with the width splitter."""
     mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
-    monitor_selection_action = mw.toolbar.components.get_action("monitor_selection")
-    cmap_action = mw.toolbar.components.get_action("color_map")
 
-    monitor_selection_action.combobox.addItem("waveform1d")
-    monitor_selection_action.combobox.setCurrentText("waveform1d")
+    device_selection = mw.toolbar.components.get_action("device_selection").widget
+    assert device_selection.device_combo_box is not None
+    assert device_selection.signal_combo_box is not None
+    # The section-width splitter of the shared component is registered as well
+    assert mw.toolbar.components.exists("device_selection_splitter")
+    # The color map widget remains available
+    assert mw.toolbar.components.get_action("color_map").widget is not None
+
+
+def test_selection_toolbar_updates_widget(qtbot, mocked_client, monkeypatch):
+    """
+    Selecting a device and signal from the toolbar comboboxes connects the widget,
+    and selecting a colormap updates the color palette.
+    """
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+
+    def fake_get(signal_class_filter):
+        signal_classes = (
+            signal_class_filter
+            if isinstance(signal_class_filter, (list, tuple, set))
+            else [signal_class_filter]
+        )
+        if "PreviewSignal" in signal_classes:
+            return [
+                (
+                    "waveform1d",
+                    "wave",
+                    {
+                        "obj_name": "waveform1d_wave",
+                        "component_name": "wave",
+                        "signal_class": "PreviewSignal",
+                        "describe": {"signal_info": {"ndim": 1}},
+                    },
+                )
+            ]
+        return []
+
+    monkeypatch.setattr(mw.client.device_manager, "get_bec_signals", fake_get)
+    device_selection = mw.toolbar.components.get_action("device_selection").widget
+    device_selection.device_combo_box.update_devices_from_filters()
+
+    device_selection.device_combo_box.setCurrentText("waveform1d")
     assert mw.monitor == "waveform1d"
 
+    # The signal combobox offers the monitor_1d sentinel and the preview signal
+    signal_items = [
+        device_selection.signal_combo_box.itemText(i)
+        for i in range(device_selection.signal_combo_box.count())
+    ]
+    assert "monitor_1d" in signal_items
+    assert "wave" in signal_items
+
+    device_selection.signal_combo_box.setCurrentText("wave")
+    assert mw.monitor_signal == "wave"
+    assert bridges[-1].scan is None
+    assert bridges[-1].sources == [("waveform1d", "wave")]
+
+    cmap_action = mw.toolbar.components.get_action("color_map")
     cmap_action.widget.colormap = "viridis"
     assert mw.color_palette == "viridis"
 
 
+def test_selection_toolbar_monitor_1d_entry(qtbot, mocked_client, monkeypatch):
+    """Selecting the monitor_1d sentinel routes to the scan-less device stream."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+
+    def fake_get(signal_class_filter):
+        return []
+
+    monkeypatch.setattr(mw.client.device_manager, "get_bec_signals", fake_get)
+    device_selection = mw.toolbar.components.get_action("device_selection").widget
+    device_selection.device_combo_box.update_devices_from_filters()
+
+    # Async-readout devices are listed even without matching class signals
+    assert "eiger" in device_selection.device_combo_box.devices
+
+    device_selection.device_combo_box.setCurrentText("eiger")
+    device_selection.signal_combo_box.setCurrentText("monitor_1d")
+    assert mw.monitor == "eiger"
+    assert mw.monitor_signal == "monitor_1d"
+    assert bridges[-1].scan is None
+    assert bridges[-1].sources == [("eiger", "monitor_1d")]
+
+
+def test_toolbar_syncs_from_properties(qtbot, mocked_client, monkeypatch):
+    """Programmatic plot() calls are mirrored into the selection comboboxes."""
+    _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+
+    mw.plot("waveform1d")
+    qtbot.wait(100)
+    device_selection = mw.toolbar.components.get_action("device_selection").widget
+    assert device_selection.device_combo_box.currentText() == "waveform1d"
+    assert device_selection.signal_combo_box.currentText() == "monitor_1d"
+
+
 def test_control_panel_opacity_slider_spinbox(qtbot, mocked_client):
     """
     Verify that when the user moves the opacity slider or spinbox, the widget's