fix(waveform): repair async signal streaming for BEC async signal curves

bec_widgets/tests/utils.py

@@ -259,7 +259,13 @@ class DMMock:
         API shape used by Waveform._check_async_signal_found.
         """
         signals: list[tuple[str, str, dict]] = []
-        if signal_class_name != "AsyncSignal":
+        # The real API accepts a single class name or a list of class names
+        # (e.g. Waveform passes sorted(ASYNC_SIGNAL_CLASSES)).
+        if isinstance(signal_class_name, str):
+            signal_class_names = [signal_class_name]
+        else:
+            signal_class_names = list(signal_class_name)
+        if "AsyncSignal" not in signal_class_names:
             return signals
 
         for device in self.devices.values():

bec_widgets/widgets/plots/waveform/settings/curve_settings/curve_tree.py

@@ -449,7 +449,7 @@ class CurveRow(QTreeWidgetItem):
                         device_entry = device_entry_info.get("obj_name", device_entry)
                 else:
                     device_entry = self.entry_validator.validate_signal(
-                        device=device_name, signal=device_entry
+                        name=device_name, entry=device_entry
                     )
 
             self.config.signal = DeviceSignal(device=device_name, signal=device_entry)

bec_widgets/widgets/plots/waveform/waveform.py

@@ -1510,13 +1510,30 @@ class Waveform(PlotBase):
         """
         Clean up the curve by disconnecting the async update signal (even for sync curves).
 
+        The endpoint is resolved the same way _setup_async_curve resolved it
+        (device_async_signal for BEC async signals, legacy device_async_readback
+        otherwise) and the subscription is only dropped once no other curve
+        shares the stream.
+
         Args:
             curve(Curve): The curve to clean up.
         """
-        self.bec_dispatcher.disconnect_slot(
-            self.on_async_readback,
-            MessageEndpoints.device_async_readback(self.scan_id, curve.name()),
-        )
+        name = getattr(curve.config.signal, "device", None) if curve.config.signal else None
+        signal = getattr(curve.config.signal, "signal", None) if curve.config.signal else None
+        if name is not None:
+            async_signal_found, stream = self._check_async_signal_found(name, signal)
+            if async_signal_found:
+                stream_key = (name, stream)
+                endpoint = MessageEndpoints.device_async_signal(self.scan_id, name, stream)
+            else:
+                stream_key = (name, None)
+                endpoint = MessageEndpoints.device_async_readback(self.scan_id, name)
+            subscribers = self._async_streams_setup.get(stream_key)
+            if subscribers is not None and signal in subscribers:
+                subscribers.remove(signal)
+            if not subscribers:
+                self._async_streams_setup.pop(stream_key, None)
+                self.bec_dispatcher.disconnect_slot(self.on_async_readback, endpoint)
         curve.rpc_register.remove_rpc(curve)
 
         # Remove itself from the DAP summary only for side panels
@@ -1700,6 +1717,15 @@ class Waveform(PlotBase):
             device_entry = curve.config.signal.signal
             if access_key == "val":  # live access
                 device_data = data.get(device_name, {}).get(device_entry, {}).get(access_key, None)
+                if device_data is None:
+                    # Live async data never passes through scan_item.live_data; it
+                    # arrives via the Redis stream subscription (on_async_readback).
+                    # Nothing to redraw here — and nothing to warn about.
+                    logger.debug(
+                        f"No live_data entry for async curve {curve.name()}; "
+                        "updates come from the async stream subscription."
+                    )
+                    continue
             else:  # history access
                 dataset_obj = data.get(device_name, {})
                 if self._skip_large_dataset_check is False:
@@ -1844,6 +1870,7 @@ class Waveform(PlotBase):
         scan_id = sender.cb_info.get("scan_id", None)
         if scan_id != self.scan_id:
             logger.info("Scan ID mismatch, ignoring async readback.")
+            return
 
         instruction = metadata.get("async_update", {}).get("type")
         if instruction not in ["add", "add_slice", "replace"]:

tests/unit_tests/test_waveform.py

@@ -8,6 +8,7 @@ from unittest.mock import MagicMock
 import numpy as np
 import pyqtgraph as pg
 import pytest
+from bec_lib.endpoints import MessageEndpoints
 from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem
 from qtpy.QtCore import QTimer
 from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox
@@ -913,6 +914,55 @@ def test_setup_async_curve(qtbot, mocked_client, monkeypatch):
     connect_spy.assert_called_once()
 
 
+def test_setup_async_curve_uses_device_async_signal_for_bec_signals(
+    qtbot, mocked_client, monkeypatch
+):
+    """
+    A curve backed by a declared BEC async signal (e.g. a DynamicSignal group
+    sub-signal) must subscribe to the per-signal device_async_signal endpoint,
+    not the legacy per-device device_async_readback stream.
+    """
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    wf.old_scan_id = "111"
+    wf.scan_id = "222"
+
+    c = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf._async_curves = [c]
+    c.config.signal.signal = "async_device_mean_ai0"
+
+    monkeypatch.setattr(
+        wf.client.device_manager,
+        "get_bec_signals",
+        lambda classes: [
+            (
+                "async_device",
+                "mean.ai0",
+                {
+                    "obj_name": "async_device_mean_ai0",
+                    "storage_name": "async_device_mean",
+                    "signal_class": "DynamicSignal",
+                },
+            )
+        ],
+    )
+    connect_spy = MagicMock()
+    monkeypatch.setattr(wf.bec_dispatcher, "connect_slot", connect_spy)
+
+    wf._async_streams_setup = {}
+    wf._setup_async_curve(c)
+    connect_spy.assert_called_once()
+    endpoint_called = connect_spy.call_args[0][1].endpoint
+    expected = MessageEndpoints.device_async_signal("222", "async_device", "async_device_mean")
+    assert endpoint_called == expected.endpoint
+
+    # cleanup drops the same endpoint once the last subscriber is removed
+    disconnect_spy = MagicMock()
+    monkeypatch.setattr(wf.bec_dispatcher, "disconnect_slot", disconnect_spy)
+    wf._curve_clean_up(c)
+    assert disconnect_spy.call_args[0][1].endpoint == expected.endpoint
+    assert (("async_device", "async_device_mean")) not in wf._async_streams_setup
+
+
 def test_on_async_readback_add_update(qtbot, mocked_client):
     """
     Test that on_async_readback extends or replaces async data depending on metadata instruction.