fix(positioner): fail move on final read errors

ophyd_devices/interfaces/base_classes/psi_positioner_base.py

@@ -8,7 +8,9 @@ from ophyd import Kind
 from ophyd.device import required_for_connection
 from ophyd.positioner import PositionerBase
 from ophyd.signal import EpicsSignalBase, Signal
+from ophyd.status import StatusTimeoutError, WaitTimeoutError
 from ophyd.status import wait as status_wait
+from ophyd.utils import UnknownStatusFailure
 from ophyd.utils.epics_pvs import AlarmSeverity, fmt_time
 
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
@@ -23,28 +25,65 @@ class MoveStatusWithTolerance(MoveStatus):
     """A MoveStatus that checks the final position against a tolerance if provided."""
 
     def __init__(self, positioner, *args, **kwargs):
-        self._tolerance_exception = None
+        self._completion_exception = None
         super().__init__(positioner, *args, **kwargs)
         self.positioner = positioner
 
     def _finished(self, success: bool = True, **kwargs):
-
-        if not success or self.positioner.user_readback is _OPTIONAL_SIGNAL:
+        if not success:
+            exc = UnknownStatusFailure(
+                f"The status {self!r} has failed. To obtain more specific, "
+                "helpful errors in the future, update the Device to use "
+                "set_exception(...) instead of _finished(success=False)."
+            )
+            return self._fail_move(exc)
+        if self.positioner.user_readback is _OPTIONAL_SIGNAL:
             return super()._finished(success=success, **kwargs)
 
-        tol = self.positioner.tolerance.get()
-        final_position = self.positioner.user_readback.get()
-        if abs(self.target - final_position) > tol:
+        try:
+            tol = self.positioner.tolerance.get()
+            skip_check = bool(tol == np.inf)
+            if not skip_check:
+                final_position = self.positioner.user_readback.get()
+                outside_tolerance = bool(abs(self.target - final_position) > tol)
+        except Exception as exc:
+            if isinstance(exc, (StatusTimeoutError, WaitTimeoutError)):
+                wrapped = RuntimeError(f"Final position check failed: {exc}")
+                wrapped.__cause__ = exc
+                exc = wrapped
+            return self._fail_move(exc, motion_completed=True)
+
+        if skip_check:
+            return super()._finished(success=success, **kwargs)
+        if outside_tolerance:
             exc = RuntimeError(
                 f"Move to {self.target} failed, "
                 f"final position {final_position} outside of tolerance {tol}"
             )
-            self._tolerance_exception = exc
-            return self.set_exception(exc)
+            return self._fail_move(exc, motion_completed=True)
         return super()._finished(success=success, **kwargs)
 
+    def _fail_move(self, exc: Exception, *, motion_completed: bool = False):
+        # The custom MoveStatus timeout is not a subclass of ophyd's
+        # StatusTimeoutError, so ophyd's set_exception() can overwrite it.
+        # Use the same lock as the timeout thread to keep the first failure.
+        with self._externally_initiated_completion_lock:
+            if self._externally_initiated_completion or self._exception is not None:
+                return
+            self._externally_initiated_completion = True
+            self._exception = exc
+            if motion_completed:
+                self._completion_exception = exc
+            self._trace_attributes["exception"] = exc
+            self._settled_event.set()
+
+        self._close_trace()
+        if self._callback_thread is None:
+            self._run_callbacks()
+
     def _handle_failure(self):
-        if self._tolerance_exception is not None and self._exception is self._tolerance_exception:
+        # Motion has already completed, so a stop here could issue another move.
+        if self._completion_exception is not None and self._exception is self._completion_exception:
             return
         super()._handle_failure()
 

tests/test_psi_positioner.py

@@ -6,6 +6,7 @@ import pytest
 from ophyd.device import Component as Cpt
 from ophyd.signal import EpicsSignal, Kind, Signal
 from ophyd.sim import FakeEpicsSignal, FakeEpicsSignalRO
+from ophyd.utils import StatusTimeoutError
 
 from ophyd_devices.devices.simple_positioner import PSISimplePositioner
 from ophyd_devices.interfaces.base_classes.psi_positioner_base import (
@@ -14,6 +15,7 @@ from ophyd_devices.interfaces.base_classes.psi_positioner_base import (
     RequiredSignalNotSpecified,
 )
 from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
+from ophyd_devices.utils.psi_device_base_utils import StatusTimeoutErrorWithErrorInfo
 
 
 def test_cannot_isntantiate_without_required_signals():
@@ -289,6 +291,129 @@ def test_tolerance_checks_requested_target_even_if_setpoint_changes(
         st.wait(timeout=1)
 
 
+@pytest.mark.parametrize("signal_name", ["tolerance", "user_readback"])
+def test_completion_read_error_fails_without_stopping_or_rewriting_setpoint(
+    mock_done_signal_positioner_with_tolerance, signal_name
+):
+    dev = mock_done_signal_positioner_with_tolerance
+    st = dev.move(5, wait=False)
+    dev.motor_done_move.sim_put(0)
+    callbacks_finished = threading.Event()
+    st.add_callback(lambda _: callbacks_finished.set())
+    read_error = RuntimeError(f"{signal_name} read failed")
+
+    with (
+        patch.object(getattr(dev, signal_name), "get", side_effect=read_error),
+        patch.object(dev, "stop", wraps=dev.stop) as stop,
+        patch.object(dev.user_setpoint, "put", wraps=dev.user_setpoint.put) as put,
+    ):
+        dev.motor_done_move.sim_put(1)
+        assert callbacks_finished.wait(timeout=1)
+        with pytest.raises(RuntimeError) as raised:
+            st.wait(timeout=1)
+        assert raised.value is read_error
+        stop.assert_not_called()
+        put.assert_not_called()
+
+
+def test_completion_read_status_timeout_is_wrapped_and_finishes(
+    mock_done_signal_positioner_with_tolerance,
+):
+    dev = mock_done_signal_positioner_with_tolerance
+    st = dev.move(5, wait=False)
+    dev.motor_done_move.sim_put(0)
+    callbacks_finished = threading.Event()
+    st.add_callback(lambda _: callbacks_finished.set())
+    read_error = StatusTimeoutError("final read timed out")
+
+    with patch.object(dev.tolerance, "get", side_effect=read_error):
+        dev.motor_done_move.sim_put(1)
+        assert callbacks_finished.wait(timeout=1)
+
+    with pytest.raises(RuntimeError) as raised:
+        st.wait(timeout=1)
+    assert raised.value.__cause__ is read_error
+
+
+def test_late_completion_read_error_preserves_move_timeout_and_stop(
+    mock_done_signal_positioner_with_tolerance,
+):
+    dev = mock_done_signal_positioner_with_tolerance
+    st = dev.move(5, wait=False, timeout=0.5)
+    dev.motor_done_move.sim_put(0)
+    entered_read = threading.Event()
+    release_read = threading.Event()
+    callbacks_finished = threading.Event()
+    st.add_callback(lambda _: callbacks_finished.set())
+    read_error = RuntimeError("late tolerance read failure")
+
+    def blocked_read():
+        entered_read.set()
+        assert release_read.wait(timeout=2)
+        raise read_error
+
+    with (
+        patch.object(dev.tolerance, "get", side_effect=blocked_read),
+        patch.object(dev, "stop", wraps=dev.stop) as stop,
+    ):
+        completion_thread = threading.Thread(target=dev.motor_done_move.sim_put, args=(1,))
+        completion_thread.start()
+        try:
+            assert entered_read.wait(timeout=1)
+            assert not st.done
+            assert callbacks_finished.wait(timeout=2)
+            with pytest.raises(StatusTimeoutErrorWithErrorInfo) as timed_out:
+                st.wait(timeout=1)
+            stop.assert_called_once()
+        finally:
+            release_read.set()
+            completion_thread.join(timeout=1)
+
+        assert not completion_thread.is_alive()
+        with pytest.raises(StatusTimeoutErrorWithErrorInfo) as after_read:
+            st.wait(timeout=1)
+        assert after_read.value is timed_out.value
+        stop.assert_called_once()
+
+
+def test_move_timeout_remains_structured_after_stop_cancels_done_status(
+    mock_done_signal_positioner_with_tolerance,
+):
+    dev = mock_done_signal_positioner_with_tolerance
+    st = dev.move(5, wait=False, timeout=0.2)
+    dev.motor_done_move.sim_put(0)
+    callbacks_finished = threading.Event()
+    st.add_callback(lambda _: callbacks_finished.set())
+
+    with (
+        patch.object(
+            dev.tolerance, "get", side_effect=AssertionError("unexpected final read")
+        ) as get,
+        patch.object(dev, "stop", wraps=dev.stop) as stop,
+        patch.object(dev, "_done_moving", wraps=dev._done_moving) as done_moving,
+    ):
+        assert callbacks_finished.wait(timeout=2)
+        with pytest.raises(StatusTimeoutErrorWithErrorInfo):
+            st.wait(timeout=1)
+        stop.assert_called_once()
+        done_moving.assert_any_call(success=False)
+        get.assert_not_called()
+
+
+def test_infinite_tolerance_completion_does_not_read_final_position(mock_psi_positioner):
+    dev = mock_psi_positioner
+    assert dev.tolerance.get() == float("inf")
+    st = dev.move(5, wait=False)
+    dev.motor_done_move._read_pv.mock_data = 0
+
+    with patch.object(dev.user_readback, "get", side_effect=RuntimeError("unexpected read")) as get:
+        dev.motor_done_move._read_pv.mock_data = 1
+        st.wait(timeout=1)
+        get.assert_not_called()
+
+    assert st.success
+
+
 def test_move_rejects_tolerance_tighter_than_deadband_before_setpoint_write(
     mock_done_signal_positioner_with_tolerance,
 ):