csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py

@@ -14,6 +14,7 @@ from prettytable import PrettyTable
 
 from csaxs_bec.devices.omny.rt.rt_ophyd import (
     RtCommunicationError,
+    RtDataMonitor,
     RtError,
     RtReadbackSignal,
     RtSetpointSignal,
@@ -520,6 +521,7 @@ class RtFlomniController(Controller):
             raise RtError("Cannot start scan because no target positions are planned.")
             # hier exception
         # start a point-by-point scan (for cont scan in flomni it would be "sa")
+        logger.info("RtFlomniController: Starting scan.")
         self.socket_put_and_receive("sd")
 
     @retry_once
@@ -809,7 +811,7 @@ class RtFlomniMotor(Device, PositionerBase):
 
 
 class RtFlomniFlyer(Device):
-    USER_ACCESS = ["controller"]
+    USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"]
     data = Cpt(
         AsyncMultiSignal,
         name="data",
@@ -846,9 +848,12 @@ class RtFlomniFlyer(Device):
         port=2222,
         socket_cls=SocketIO,
         device_manager=None,
+        data_timeout=5.0,
+        scan_timeout=60.0,
         **kwargs,
     ):
         super().__init__(prefix=prefix, name=name, parent=parent, **kwargs)
+        self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout)
         self.shutdown_event = threading.Event()
         self.controller = RtFlomniController(
             socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
@@ -859,6 +864,49 @@ class RtFlomniFlyer(Device):
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
 
+    @property
+    def data_timeout(self) -> float:
+        """Seconds without received data before logging diagnostics (default: 5)."""
+        return self._data_monitor.data_timeout
+
+    @data_timeout.setter
+    def data_timeout(self, value: float) -> None:
+        self._data_monitor.data_timeout = value
+
+    @property
+    def scan_timeout(self) -> float:
+        """Seconds without received points before failing the scan (default: 60).
+
+        Includes initial buffering: during an active scan, samples are only read
+        once the reported scan position exceeds five. Allow enough time for this
+        first read; each received sample then resets the timeout.
+        """
+        return self._data_monitor.scan_timeout
+
+    @scan_timeout.setter
+    def scan_timeout(self, value: float) -> None:
+        self._data_monitor.scan_timeout = value
+
+    def fetch_diagnostics(self) -> dict:
+        """Return scan status, feedback state, and recent controller commands."""
+        diagnostics = {}
+        try:
+            mode, planned, current = self.controller.get_scan_status()
+            diagnostics["scan_status"] = {
+                "mode": mode,
+                "planned_positions": planned,
+                "current_position": current,
+            }
+        except Exception as exc:
+            diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}"
+        try:
+            diagnostics["feedback_running"] = self.controller.feedback_is_running()
+        except Exception as exc:
+            diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}"
+        with self.controller._lock:
+            diagnostics["command_history"] = list(self.controller.command_history)
+        return diagnostics
+
     def read_positions_from_sampler(self, status: DeviceStatus):
         """
         Read the positions from the sampler and update the data signal.
@@ -894,6 +942,13 @@ class RtFlomniFlyer(Device):
                     read_counter = read_counter + 1
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
+                    self._data_monitor.reset()
+
+            if mode > 0 and not self.shutdown_event.is_set():
+                exception = self._data_monitor.check(self.name, self.fetch_diagnostics)
+                if exception is not None:
+                    status.set_exception(exception)
+                    return
 
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
@@ -910,6 +965,7 @@ class RtFlomniFlyer(Device):
 
             signals = self._get_signals_from_table(return_table)
             self.data.set(signals)
+            self._data_monitor.reset()
 
         # NOTE: No need to set the status to failed if the shutdown_event is set.
         # The stop() method will take care of that.
@@ -959,6 +1015,7 @@ class RtFlomniFlyer(Device):
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
+        self._data_monitor.reset()
         self.controller.start_scan()
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
@@ -980,7 +1037,10 @@ class RtFlomniFlyer(Device):
     def stop(self, *, success=False):
         self.shutdown_event.set()
         self.scan_done_event.set()
-        if self.readout_thread is not None:
+        if (
+            self.readout_thread is not None
+            and self.readout_thread is not threading.current_thread()
+        ):
             self.readout_thread.join()
         return super().stop(success=success)
 
@@ -990,4 +1050,4 @@ if __name__ == "__main__":
         socket_cls=SocketIO, socket_host="mpc2844.psi.ch", socket_port=2222, device_manager=None
     )
     rtcontroller.on()
-    rtcontroller.laser_tracker_on()
+    rtcontroller.laser_tracker_on()

csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py

@@ -13,7 +13,7 @@ from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
 from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
 from prettytable import PrettyTable
 
-from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtError
+from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtDataMonitor, RtError
 
 logger = bec_logger.logger
 
@@ -217,6 +217,7 @@ class RtLamniController(Controller):
             raise RtError("Cannot start scan because no target positions are planned.")
             # hier exception
         # start a point-by-point scan (for cont scan in flomni it would be "sa")
+        logger.info("RtLamniController: Starting scan.")
         self.socket_put_and_receive("sd")
 
     def feedback_is_running(self) -> bool:
@@ -695,7 +696,7 @@ class RtLamniMotor(Device, PositionerBase):
 
 
 class RtLamniFlyer(Device):
-    USER_ACCESS = ["controller"]
+    USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"]
     data = Cpt(
         AsyncMultiSignal,
         name="data",
@@ -741,9 +742,12 @@ class RtLamniFlyer(Device):
         port=3333,
         socket_cls=SocketIO,
         device_manager=None,
+        data_timeout=5.0,
+        scan_timeout=60.0,
         **kwargs,
     ):
         super().__init__(prefix=prefix, name=name, parent=parent, **kwargs)
+        self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout)
         self.shutdown_event = threading.Event()
         self.controller = RtLamniController(
             socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
@@ -755,6 +759,49 @@ class RtLamniFlyer(Device):
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
 
+    @property
+    def data_timeout(self) -> float:
+        """Seconds without received data before logging diagnostics (default: 5)."""
+        return self._data_monitor.data_timeout
+
+    @data_timeout.setter
+    def data_timeout(self, value: float) -> None:
+        self._data_monitor.data_timeout = value
+
+    @property
+    def scan_timeout(self) -> float:
+        """Seconds without received points before failing the scan (default: 60).
+
+        Includes initial buffering: during an active scan, samples are only read
+        once the reported scan position exceeds five. Allow enough time for this
+        first read; each received sample then resets the timeout.
+        """
+        return self._data_monitor.scan_timeout
+
+    @scan_timeout.setter
+    def scan_timeout(self, value: float) -> None:
+        self._data_monitor.scan_timeout = value
+
+    def fetch_diagnostics(self) -> dict:
+        """Return scan status, feedback state, and recent controller commands."""
+        diagnostics = {}
+        try:
+            mode, planned, current = self.controller.get_scan_status()
+            diagnostics["scan_status"] = {
+                "mode": mode,
+                "planned_positions": planned,
+                "current_position": current,
+            }
+        except Exception as exc:
+            diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}"
+        try:
+            diagnostics["feedback_running"] = self.controller.feedback_is_running()
+        except Exception as exc:
+            diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}"
+        with self.controller._lock:
+            diagnostics["command_history"] = list(self.controller.command_history)
+        return diagnostics
+
     def stage(self):
         self.shutdown_event.clear()
         self.scan_done_event.set()
@@ -771,6 +818,7 @@ class RtLamniFlyer(Device):
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
+        self._data_monitor.reset()
         self.controller.start_scan()
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
@@ -792,7 +840,10 @@ class RtLamniFlyer(Device):
     def stop(self, *, success=False):
         self.shutdown_event.set()
         self.scan_done_event.set()
-        if self.readout_thread is not None:
+        if (
+            self.readout_thread is not None
+            and self.readout_thread is not threading.current_thread()
+        ):
             self.readout_thread.join()
         return super().stop(success=success)
 
@@ -832,6 +883,13 @@ class RtLamniFlyer(Device):
                     read_counter = read_counter + 1
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
+                    self._data_monitor.reset()
+
+            if mode > 0 and not self.shutdown_event.is_set():
+                exception = self._data_monitor.check(self.name, self.fetch_diagnostics)
+                if exception is not None:
+                    status.set_exception(exception)
+                    return
 
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
@@ -848,6 +906,7 @@ class RtLamniFlyer(Device):
 
             signals = self._get_signals_from_table(return_table)
             self.data.set(signals)
+            self._data_monitor.reset()
 
         # NOTE: No need to set the status to failed if the shutdown_event is set.
         # The stop() method will take care of that.

csaxs_bec/devices/omny/rt/rt_omny_ophyd.py

@@ -23,6 +23,7 @@ if builtins.__dict__.get("bec") is not None:
 
 from csaxs_bec.devices.omny.rt.rt_ophyd import (
     RtCommunicationError,
+    RtDataMonitor,
     RtError,
     RtReadbackSignal,
     RtSetpointSignal,
@@ -859,6 +860,7 @@ class RtOMNYController(Controller):
             raise RtError("Cannot start scan because no target positions are planned.")
             # hier exception
         # start a point-by-point scan (for cont scan in flomni it would be "sa")
+        logger.info("RtOMNYController: Starting scan.")
         self.socket_put_and_receive("sd")
 
     @retry_once
@@ -1140,7 +1142,7 @@ class RtOMNYMotor(Device, PositionerBase):
 
 
 class RtOMNYFlyer(Device):
-    USER_ACCESS = ["controller"]
+    USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"]
 
     data = Cpt(
         AsyncMultiSignal,
@@ -1184,9 +1186,12 @@ class RtOMNYFlyer(Device):
         port=2222,
         socket_cls=SocketIO,
         device_manager=None,
+        data_timeout=5.0,
+        scan_timeout=60.0,
         **kwargs,
     ):
         super().__init__(prefix=prefix, name=name, parent=parent, **kwargs)
+        self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout)
         self.shutdown_event = threading.Event()
         self.controller = RtOMNYController(
             socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
@@ -1197,6 +1202,49 @@ class RtOMNYFlyer(Device):
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
 
+    @property
+    def data_timeout(self) -> float:
+        """Seconds without received data before logging diagnostics (default: 5)."""
+        return self._data_monitor.data_timeout
+
+    @data_timeout.setter
+    def data_timeout(self, value: float) -> None:
+        self._data_monitor.data_timeout = value
+
+    @property
+    def scan_timeout(self) -> float:
+        """Seconds without received points before failing the scan (default: 60).
+
+        Includes initial buffering: during an active scan, samples are only read
+        once the reported scan position exceeds five. Allow enough time for this
+        first read; each received sample then resets the timeout.
+        """
+        return self._data_monitor.scan_timeout
+
+    @scan_timeout.setter
+    def scan_timeout(self, value: float) -> None:
+        self._data_monitor.scan_timeout = value
+
+    def fetch_diagnostics(self) -> dict:
+        """Return scan status, feedback state, and recent controller commands."""
+        diagnostics = {}
+        try:
+            mode, planned, current = self.controller.get_scan_status()
+            diagnostics["scan_status"] = {
+                "mode": mode,
+                "planned_positions": planned,
+                "current_position": current,
+            }
+        except Exception as exc:
+            diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}"
+        try:
+            diagnostics["feedback_running"] = self.controller.feedback_is_running()
+        except Exception as exc:
+            diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}"
+        with self.controller._lock:
+            diagnostics["command_history"] = list(self.controller.command_history)
+        return diagnostics
+
     def stage(self):
         self.shutdown_event.clear()
         self.scan_done_event.set()
@@ -1213,6 +1261,7 @@ class RtOMNYFlyer(Device):
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
+        self._data_monitor.reset()
         self.controller.start_scan()
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
@@ -1234,7 +1283,10 @@ class RtOMNYFlyer(Device):
     def stop(self, *, success=False):
         self.shutdown_event.set()
         self.scan_done_event.set()
-        if self.readout_thread is not None:
+        if (
+            self.readout_thread is not None
+            and self.readout_thread is not threading.current_thread()
+        ):
             self.readout_thread.join()
         return super().stop(success=success)
 
@@ -1251,7 +1303,9 @@ class RtOMNYFlyer(Device):
         self.average_stdeviations_x_st_fzp = 0
         self.average_stdeviations_y_st_fzp = 0
 
-        mode, number_of_positions_planned, current_position_in_scan = self.get_scan_status()
+        mode, number_of_positions_planned, current_position_in_scan = (
+            self.controller.get_scan_status()
+        )
         while mode > 0 and not self.shutdown_event.wait(0.01):
             # logger.info(f"Current scan position {current_position_in_scan} out of {number_of_positions_planned}")
             mode, number_of_positions_planned, current_position_in_scan = (
@@ -1269,6 +1323,13 @@ class RtOMNYFlyer(Device):
                     read_counter = read_counter + 1
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
+                    self._data_monitor.reset()
+
+            if mode > 0 and not self.shutdown_event.is_set():
+                exception = self._data_monitor.check(self.name, self.fetch_diagnostics)
+                if exception is not None:
+                    status.set_exception(exception)
+                    return
 
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
@@ -1285,6 +1346,7 @@ class RtOMNYFlyer(Device):
 
             signals = self._get_signals_from_table(return_table)
             self.data.set(signals)
+            self._data_monitor.reset()
 
         # NOTE: No need to set the status to failed if the shutdown_event is set.
         # The stop() method will take care of that.

csaxs_bec/devices/omny/rt/rt_ophyd.py

@@ -1,10 +1,12 @@
 """
-This module contains base signals for RT devices. Controller and motors are implemented in the
-bespoke modules such as `rt_flomni_ophyd.py` or `rt_lamni_ophyd.py`.
+This module contains base signals and data monitoring for RT devices. Controllers and motors are
+implemented in bespoke modules such as `rt_flomni_ophyd.py` or `rt_lamni_ophyd.py`.
 """
 
 import functools
+import math
 import time
+from collections.abc import Callable
 
 from bec_lib import bec_logger
 from ophyd.utils import ReadOnlyError
@@ -26,6 +28,82 @@ class BECConfigError(Exception):
     pass
 
 
+class RtDataMonitor:
+    """Track gaps in received scan samples for an owning RT flyer.
+
+    ``data_timeout`` warns once with diagnostics; ``scan_timeout`` fails the scan.
+    The readout loop checks both after draining available samples. Status replies do
+    not count as data, and receiving a sample resets both deadlines. This does not
+    monitor blocked socket I/O.
+
+    Both intervals start at kickoff and include initial sampler buffering. While
+    the scan is active, RT flyers only fetch samples once the reported scan
+    position exceeds five. Controller progress alone does not reset either timer,
+    so slow scans may warn or time out before this first batch is read. Configure
+    ``scan_timeout`` to allow enough time for the first sampler read. It limits
+    gaps between received samples, not the total scan duration.
+    """
+
+    def __init__(self, data_timeout=5.0, scan_timeout=60.0):
+        self.data_timeout = data_timeout
+        self.scan_timeout = scan_timeout
+        self._last_data_time = None
+        self._data_timeout_logged = False
+
+    @staticmethod
+    def _validate_timeout(value, name):
+        value = float(value)
+        if not math.isfinite(value) or value <= 0:
+            raise ValueError(f"{name} must be a finite, positive number of seconds.")
+        return value
+
+    @property
+    def data_timeout(self) -> float:
+        """Seconds without a received sample before logging a warning; must be positive."""
+        return self._data_timeout
+
+    @data_timeout.setter
+    def data_timeout(self, value: float) -> None:
+        self._data_timeout = self._validate_timeout(value, "data_timeout")
+
+    @property
+    def scan_timeout(self) -> float:
+        """Seconds without a received sample before failing the scan."""
+        return self._scan_timeout
+
+    @scan_timeout.setter
+    def scan_timeout(self, value: float) -> None:
+        self._scan_timeout = self._validate_timeout(value, "scan_timeout")
+
+    def reset(self) -> None:
+        """Start a new interval at kickoff or after receiving a sample."""
+        self._last_data_time = time.monotonic()
+        self._data_timeout_logged = False
+
+    def check(self, device_name: str, fetch_diagnostics: Callable[[], dict]) -> TimeoutError | None:
+        """Log a warning or return the exception the flyer should set on its status."""
+        if self._last_data_time is None:
+            return
+        elapsed = time.monotonic() - self._last_data_time
+        if elapsed >= self.scan_timeout:
+            return TimeoutError(
+                f"No RT scan points received by {device_name} for {elapsed:.1f} s "
+                f"(scan_timeout={self.scan_timeout:g} s)."
+            )
+        if elapsed < self.data_timeout or self._data_timeout_logged:
+            return
+        # Latch before fetching diagnostics so a failed diagnostic cannot cause log spam.
+        self._data_timeout_logged = True
+        try:
+            diagnostics = fetch_diagnostics()
+        except Exception as exc:
+            diagnostics = {"error": f"{type(exc).__name__}: {exc}"}
+        logger.warning(
+            f"No RT scan data received by {device_name} for {elapsed:.1f} s "
+            f"(data_timeout={self.data_timeout:g} s). Diagnostics: {diagnostics}"
+        )
+
+
 class RtSignalBase(SocketSignal):
     def __init__(self, signal_name, **kwargs):
         self.signal_name = signal_name

csaxs_bec/scans/flomni_fermat_scan.py

@@ -232,6 +232,7 @@ class FlomniFermatScan(ScanBase):
         status = self.actions.complete(device=self.dev.rt_positions, wait=False)
         while not status.done:
             self.at_each_point()
+        status.wait()
 
     @scan_hook
     def at_each_point(self):

csaxs_bec/scans/lamni_fermat_scan.py

@@ -226,6 +226,7 @@ class LamniFermatScan(ScanBase):
         while not status.done:
             self.at_each_point()
             time.sleep(1)
+        status.wait()
 
     @scan_hook
     def at_each_point(self):

csaxs_bec/scans/omny_fermat_scan.py

@@ -207,6 +207,7 @@ class OmnyFermatScan(ScanBase):
         while not status.done:
             self.at_each_point()
             time.sleep(1)
+        status.wait()
 
     @scan_hook
     def at_each_point(self):

tests/tests_devices/test_rt_data_monitor.py

@@ -0,0 +1,214 @@
+"""No-data warnings must survive diagnostics failures and rearm when samples resume."""
+
+import importlib
+import threading
+from collections import deque
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+from csaxs_bec.devices.omny.rt import rt_ophyd
+
+
+@pytest.fixture(params=[("flomni", "RtFlomni"), ("lamni", "RtLamni"), ("omny", "RtOMNY")])
+def flyer(request, monkeypatch):
+    suffix, class_prefix = request.param
+    module = importlib.import_module(f"csaxs_bec.devices.omny.rt.rt_{suffix}_ophyd")
+    controller = mock.Mock()
+    controller._lock = threading.RLock()
+    controller._min_scan_buffer_reached = True
+    controller.command_history = deque(["[PUT]: sd", "[GET]: scan started"])
+    controller.get_scan_status.return_value = (2, 10, 6)
+    controller.feedback_is_running.return_value = True
+    controller.socket_put_and_receive.side_effect = lambda command: ",".join(
+        [command[1:]] + ["0"] * 24
+    )
+    monkeypatch.setattr(module, f"{class_prefix}Controller", lambda **kwargs: controller)
+    device = getattr(module, f"{class_prefix}Flyer")(name="rt_positions", data_timeout=3)
+    monkeypatch.setattr(device.data, "set", mock.Mock())
+    monkeypatch.setattr(device.progress, "put", mock.Mock())
+    yield device
+    device.destroy()
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    clock = SimpleNamespace(now=100)
+    monkeypatch.setattr(rt_ophyd, "time", SimpleNamespace(monotonic=lambda: clock.now))
+    return clock
+
+
+def run_readout(flyer, clock, states, *, finish=False, failure=False):
+    """Poll scripted (elapsed seconds, mode, position) without sleeping or hardware."""
+    origin = clock.now
+    states = iter(states)
+
+    def next_status():
+        try:
+            elapsed, mode, position = next(states)
+        except StopIteration:
+            flyer.shutdown_event.set()
+            return (2, 10, 0)
+        clock.now = origin + elapsed
+        return mode, 10, position
+
+    flyer.controller.get_scan_status.side_effect = next_status
+    flyer.shutdown_event = mock.Mock(wraps=threading.Event())
+    flyer.shutdown_event.wait.side_effect = lambda timeout: flyer.shutdown_event.is_set()
+    status = flyer.kickoff()
+    assert status.done and status.success
+    status = flyer.complete()
+    flyer.readout_thread.join(timeout=2)
+    assert not flyer.readout_thread.is_alive()
+    if failure:
+        with pytest.raises(TimeoutError, match="scan_timeout"):
+            status.wait(timeout=1)
+        assert status.done and not status.success
+    else:
+        status.wait(timeout=1)
+        assert status.success
+    if finish:
+        assert not flyer.shutdown_event.is_set()
+    return status
+
+
+@pytest.mark.parametrize("mode", [2, 3])
+def test_warns_once_at_configured_timeout_without_samples(flyer, clock, mode):
+    with (
+        mock.patch.object(
+            flyer, "fetch_diagnostics", return_value={"feedback_running": False}
+        ) as fetch,
+        mock.patch.object(rt_ophyd.logger, "warning") as warning,
+    ):
+        run_readout(flyer, clock, [(0, mode, 0), (2.999, mode, 1)])
+        warning.assert_not_called()
+        fetch.assert_not_called()
+
+        run_readout(flyer, clock, [(0, mode, 0), (3, mode, 1), (4, mode, 2), (20, mode, 2)])
+        warning.assert_called_once()
+        fetch.assert_called_once_with()
+        assert "data_timeout=3 s" in warning.call_args.args[0]
+        assert "'feedback_running': False" in warning.call_args.args[0]
+        flyer.data.set.assert_not_called()
+
+
+def test_samples_rearm_warning_and_new_scan_resets_timeout(flyer, clock):
+    with (
+        mock.patch.object(flyer, "fetch_diagnostics", return_value={}) as fetch,
+        mock.patch.object(rt_ophyd.logger, "warning") as warning,
+    ):
+        run_readout(
+            flyer, clock, [(0, 2, 0), (3, 2, 0), (4, 2, 6), (6.999, 2, 6), (7, 2, 6), (20, 2, 6)]
+        )
+        assert flyer.data.set.call_count == 5
+        assert warning.call_count == fetch.call_count == 2
+        warning.reset_mock()
+        run_readout(flyer, clock, [(0, 2, 0), (2.999, 2, 0)])
+        warning.assert_not_called()
+        run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0)])
+        warning.assert_called_once()
+
+
+def test_available_samples_are_read_before_timeout_check(flyer, clock):
+    with mock.patch.object(rt_ophyd.logger, "warning") as warning:
+        run_readout(flyer, clock, [(0, 2, 0), (4, 2, 6), (6.999, 2, 6)])
+    warning.assert_not_called()
+    assert flyer.data.set.call_count == 5
+
+
+def test_finished_scan_drains_samples_without_stall_warning(flyer, clock):
+    with mock.patch.object(rt_ophyd.logger, "warning") as warning:
+        run_readout(flyer, clock, [(0, 2, 0), (70, 0, 3)], finish=True)
+    warning.assert_not_called()
+    assert flyer.data.set.call_count == 10
+
+
+def test_diagnostics_include_status_feedback_and_history(flyer):
+    assert flyer.fetch_diagnostics() == {
+        "scan_status": {"mode": 2, "planned_positions": 10, "current_position": 6},
+        "feedback_running": True,
+        "command_history": ["[PUT]: sd", "[GET]: scan started"],
+    }
+
+
+def test_diagnostics_preserve_other_fields_when_a_query_fails(flyer):
+    flyer.controller.get_scan_status.side_effect = TimeoutError("status unavailable")
+    diagnostics = flyer.fetch_diagnostics()
+    assert diagnostics["scan_status_error"] == "TimeoutError: status unavailable"
+    assert diagnostics["feedback_running"] is True
+    assert diagnostics["command_history"]
+    flyer.controller.feedback_is_running.side_effect = RuntimeError("feedback unavailable")
+    assert flyer.fetch_diagnostics()["feedback_error"] == "RuntimeError: feedback unavailable"
+
+
+def test_failed_diagnostics_do_not_abort_or_repeat_warning(flyer, clock):
+    with (
+        mock.patch.object(
+            flyer, "fetch_diagnostics", side_effect=RuntimeError("diagnostic failure")
+        ),
+        mock.patch.object(rt_ophyd.logger, "warning") as warning,
+    ):
+        run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0), (20, 2, 0)])
+    warning.assert_called_once()
+    assert "RuntimeError: diagnostic failure" in warning.call_args.args[0]
+
+
+def test_stopped_device_does_not_warn(flyer, clock):
+    flyer._data_monitor.reset()
+    clock.now += 10
+    flyer.shutdown_event.set()
+    with mock.patch.object(flyer, "fetch_diagnostics") as fetch:
+        status = mock.Mock()
+        flyer.read_positions_from_sampler(status)
+        status.set_finished.assert_called_once_with()
+    fetch.assert_not_called()
+
+
+@pytest.mark.parametrize("mode", [2, 3])
+def test_scan_timeout_fails_status_without_stopping_axes(flyer, clock, mode):
+    flyer.scan_timeout = 8
+    with (
+        mock.patch.object(flyer, "fetch_diagnostics", return_value={}),
+        mock.patch.object(rt_ophyd.logger, "warning") as warning,
+    ):
+        status = run_readout(
+            flyer, clock, [(0, mode, 0), (3, mode, 0), (7.999, mode, 0), (8, mode, 0)], failure=True
+        )
+    warning.assert_called_once()
+    assert "scan_timeout=8 s" in str(status.exception())
+    assert flyer.shutdown_event.is_set()
+    assert flyer.scan_done_event.is_set()
+    flyer.controller.stop_all_axes.assert_not_called()
+
+
+def test_received_points_reset_scan_timeout(flyer, clock):
+    flyer.scan_timeout = 8
+    with mock.patch.object(flyer, "fetch_diagnostics", return_value={}):
+        status = run_readout(
+            flyer, clock, [(0, 2, 0), (7, 2, 6), (14.999, 2, 6), (15, 2, 6)], failure=True
+        )
+    assert "for 8.0 s" in str(status.exception())
+    assert flyer.data.set.call_count == 5
+
+
+def test_default_scan_timeout_is_sixty_seconds(flyer, clock):
+    assert flyer.scan_timeout == 60
+    with mock.patch.object(flyer, "fetch_diagnostics", return_value={}):
+        status = run_readout(flyer, clock, [(0, 2, 0), (59.999, 2, 0), (60, 2, 0)], failure=True)
+    assert "scan_timeout=60 s" in str(status.exception())
+
+
+def test_diagnostic_failure_does_not_prevent_scan_timeout(flyer, clock):
+    flyer.scan_timeout = 8
+    with mock.patch.object(
+        flyer, "fetch_diagnostics", side_effect=RuntimeError("diagnostic failure")
+    ):
+        run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0), (8, 2, 0)], failure=True)
+
+
+@pytest.mark.parametrize("name", ["data_timeout", "scan_timeout"])
+@pytest.mark.parametrize("timeout", [0, -1, float("nan"), float("inf")])
+def test_invalid_timeouts_are_rejected(name, timeout):
+    with pytest.raises(ValueError, match="finite, positive"):
+        rt_ophyd.RtDataMonitor(**{name: timeout})

tests/tests_scans/test_rt_scan_timeout.py

@@ -0,0 +1,44 @@
+"""RT completion errors must propagate out of the scan's polling loop."""
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from bec_lib import messages
+from bec_server.scan_server.errors import DeviceInstructionError
+from bec_server.scan_server.scan_stubs import ScanStubStatus
+
+from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
+from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan
+from csaxs_bec.scans.omny_fermat_scan import OmnyFermatScan
+
+
+@pytest.mark.parametrize("scan_cls", [FlomniFermatScan, LamniFermatScan, OmnyFermatScan])
+@pytest.mark.parametrize("failed", [False, True])
+def test_rt_scan_checks_completed_readout_result(scan_cls, failed):
+    status = ScanStubStatus(instruction_handler=mock.Mock())
+    if failed:
+        status.set_failed(
+            messages.ErrorInfo(
+                error_message="No RT scan points received within scan_timeout",
+                compact_error_message="RT scan timed out",
+                exception_type="TimeoutError",
+                device="rt_positions",
+            )
+        )
+    else:
+        status.set_done()
+    scan = SimpleNamespace(
+        actions=mock.Mock(),
+        dev=SimpleNamespace(rt_positions="rt_positions"),
+        at_each_point=mock.Mock(),
+        _scan_modifier=None,
+        _scan_modifier_hooks={},
+    )
+    scan.actions.complete.return_value = status
+    if failed:
+        with pytest.raises(DeviceInstructionError, match="RT scan timed out"):
+            scan_cls.scan_core(scan)
+    else:
+        scan_cls.scan_core(scan)
+    scan.actions.complete.assert_called_once_with(device="rt_positions", wait=False)