feat(eiger): make the missing-packet tolerance per-detector and switchable

The JungfrauJoch broker reports packet loss during data collection as a plain
error, which fails the scan. Since 2026-09-03 that error has been suppressed on
the beamline by an uncommitted edit, through a hardcoded flag on the Eiger base
class: it applied to every Eiger at once, it could be reached neither from the
client nor from deviceConfig, and it left nothing in the log.

The tolerance is now a real parameter, raise_on_missing_packets:

- named in Eiger, Eiger9M and Eiger1_5M, so that a deviceConfig key actually
  reaches the device. bec_server intersects config keys with the named
  parameters of the class, so a flag reachable only through **kwargs is
  silently dropped -- the same trap as readout_time (f450f29) and prefix
  (10be2b5). A test pins the signatures.
- exposed through USER_ACCESS as get_/set_raise_on_missing_packets, so a
  beamtime can change its mind without a redeployment. Like every runtime
  value it is shared between clients and does not survive a server restart;
  deviceConfig is what makes a choice stick.
- counted, and logged at warning level whenever an error is let through, so
  that "which scans were affected?" has an answer. get_missing_packet_events()
  returns the count.

What is tolerated is narrower than it looks: the frame-count check below still
raises when statistics.images_collected falls short of the trigger count. Only
"the broker flagged packet loss but delivered the expected number of images"
gets through, and a test pins that a short acquisition still fails.

The default stays False, i.e. tolerate, so the running beamtime is unaffected.
It should become True once the 9M's packet loss is understood, with
raise_on_missing_packets: false in that detector's deviceConfig if it still
needs it. That is one constant to change, RAISE_ON_MISSING_PACKETS.

The wording of the broker message is the only handle available, as there is no
error code for it. If JungfrauJoch rephrases it the match stops working and the
error raises again, which is the safe direction to fail in.

test_eiger_on_complete_error_message was skipped as failing "because the error
should be skipped for now due to HW issues". With the tolerance scoped to the
missing-packet message it passes again, and is no longer skipped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

csaxs_bec/devices/jungfraujoch/eiger.py

@@ -84,6 +84,19 @@ logger = bec_logger.logger
 # the trigger period allows.
 MIN_EXP_TIME = 500e-6  # s
 EIGER_READOUT_TIME = 2e-4  # s -- matches DDG2 DEFAULT_READOUT_TIMES["ab"]
+
+# NOTE The JungfrauJoch broker reports packet loss as a plain error with no error code,
+# so the only way to recognise it is the wording of its status message. If JFJoch ever
+# rephrases this, the match stops working and the error raises again -- which is the safe
+# direction to fail in.
+MISSING_PACKETS_MESSAGE = "Missing packets in data collection"
+
+# NOTE Default for raise_on_missing_packets, i.e. whether a missing-packet error from the
+# broker fails the scan. False while the 9M is losing packets during the 2026-09 beamtime
+# and scans must not die on it. Set it back to True once that is resolved, and put
+# "raise_on_missing_packets: false" in the deviceConfig of the detector that still needs
+# it -- per detector, rather than for every Eiger at once.
+RAISE_ON_MISSING_PACKETS = False
 # NOTE Each model declares its own value and passes it here; that duplication is
 # deliberate, since readout is a property of the detector (a 9M has more modules than a
 # 1.5M, and the Falcon needs 3 ms). They all happen to be 2e-4 today only because no
@@ -106,6 +119,8 @@ class Eiger(PSIDeviceBase):
         port (int): Port of the Jungfrau Joch server.
         scan_info (ScanInfo): The scan info to use.
         device_manager (DeviceManagerDS): The device manager to use.
+        raise_on_missing_packets (bool): Whether a 'missing packets' error from the
+            JungfrauJoch broker fails the scan. See RAISE_ON_MISSING_PACKETS.
         **kwargs: Additional keyword arguments.
     """
 
@@ -115,6 +130,9 @@ class Eiger(PSIDeviceBase):
         "print_detector_distance_and_center",
         "set_azi_settings",
         "get_azi_settings",
+        "get_raise_on_missing_packets",
+        "set_raise_on_missing_packets",
+        "get_missing_packet_events",
     ]
 
     file_event = Cpt(FileEventSignal, name="file_event")
@@ -130,6 +148,7 @@ class Eiger(PSIDeviceBase):
         beam_center: tuple[int, int] = (0, 0),
         scan_info: ScanInfo = None,
         readout_time: float = EIGER_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         device_manager=None,
         **kwargs,
     ):
@@ -153,6 +172,47 @@ class Eiger(PSIDeviceBase):
         if self.device_manager is not None:
             self.device_manager: DeviceManagerDS
         self.scan_parameters: ScanServerScanInfo | None = None
+        self._raise_on_missing_packets = bool(raise_on_missing_packets)
+        self._missing_packet_events = 0
+
+    def get_raise_on_missing_packets(self) -> bool:
+        """
+        Whether a 'missing packets' error from the JungfrauJoch broker fails the scan.
+
+        Returns:
+            bool: True if missing packets raise an EigerError, False if they are tolerated.
+        """
+        return self._raise_on_missing_packets
+
+    def set_raise_on_missing_packets(self, value: bool) -> None:
+        """
+        Set whether a 'missing packets' error from the JungfrauJoch broker fails the scan.
+
+        NOTE This is shared by every client of this device server, and it does NOT survive a
+        server restart: '__init__' runs again and restores the value from deviceConfig. To make
+        a choice stick for a beamtime, set 'raise_on_missing_packets' in the deviceConfig of the
+        detector instead.
+
+        Args:
+            value (bool): True to raise on missing packets, False to tolerate them.
+        """
+        self._raise_on_missing_packets = bool(value)
+        logger.warning(
+            f"Device {self.name}: raise_on_missing_packets is now "
+            f"{self._raise_on_missing_packets}."
+        )
+
+    def get_missing_packet_events(self) -> int:
+        """
+        How many missing-packet errors have been tolerated since this device was created.
+
+        Reset by a server restart, like the flag itself. Non-zero means acquisitions completed
+        that the broker had flagged, so the data of those scans may be incomplete.
+
+        Returns:
+            int: Number of tolerated missing-packet errors.
+        """
+        return self._missing_packet_events
 
     def _preview_callback(self, message: dict) -> None:
         """
@@ -477,13 +537,28 @@ class Eiger(PSIDeviceBase):
                         logger.info(
                             f"JungfrauJoch broker status: {yaml.dump(broker_status.to_dict(), indent=4)}"
                         )
-                    if broker_status.message_severity == "error":  # Raise on error
-                        raise EigerError(
-                            f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
-                        )
-                        # logger.warning(
-                        #     f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
-                        # )
+                    if broker_status.message_severity == "error":
+                        # NOTE Missing packets can be tolerated per detector, because the frame
+                        # count is still checked below: what is being let through here is "the
+                        # broker flagged packet loss but delivered the expected number of
+                        # images", not a short acquisition.
+                        if (
+                            MISSING_PACKETS_MESSAGE in (broker_status.message or "")
+                            and not self._raise_on_missing_packets
+                        ):
+                            self._missing_packet_events += 1
+                            logger.warning(
+                                f"Device {self.name} reported missing packets and the error is "
+                                f"being tolerated (raise_on_missing_packets is False), so the "
+                                f"data of this scan may be incomplete. That is tolerated event "
+                                f"{self._missing_packet_events} for this device since it was "
+                                f"created. Broker status: "
+                                f"{yaml.dump(broker_status.to_dict(), indent=4)}"
+                            )
+                        else:
+                            raise EigerError(
+                                f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
+                            )
                     # Call API endpoint to get statistics
                     statistics: MeasurementStatistics = (
                         self.jfj_client.api.statistics_data_collection_get(_request_timeout=5)

csaxs_bec/devices/jungfraujoch/eiger_1_5m.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from csaxs_bec.devices.jungfraujoch.eiger import Eiger
+from csaxs_bec.devices.jungfraujoch.eiger import RAISE_ON_MISSING_PACKETS, Eiger
 
 # Gap this model needs between the end of an acquisition and the next trigger. Kept
 # per model on purpose -- readout is a detector property, not a beamline constant. The
@@ -45,12 +45,14 @@ class Eiger1_5M(Eiger):
         scan_info: ScanInfo = None,
         device_manager: DeviceManagerDS = None,
         readout_time: float = EIGER1_5M_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         **kwargs,
     ) -> None:
         super().__init__(
             name=name,
             detector_name=DETECTOR_NAME,
             readout_time=readout_time,
+            raise_on_missing_packets=raise_on_missing_packets,
             detector_distance=detector_distance,
             beam_center=beam_center,
             scan_info=scan_info,

csaxs_bec/devices/jungfraujoch/eiger_9m.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from csaxs_bec.devices.jungfraujoch.eiger import Eiger
+from csaxs_bec.devices.jungfraujoch.eiger import RAISE_ON_MISSING_PACKETS, Eiger
 
 if TYPE_CHECKING:  # pragma no cover
     from bec_lib.devicemanager import ScanInfo
@@ -49,12 +49,14 @@ class Eiger9M(Eiger):
         scan_info: ScanInfo = None,
         device_manager: DeviceManagerDS = None,
         readout_time: float = EIGER9M_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         **kwargs,
     ) -> None:
         super().__init__(
             name=name,
             detector_name=DETECTOR_NAME,
             readout_time=readout_time,
+            raise_on_missing_packets=raise_on_missing_packets,
             detector_distance=detector_distance,
             beam_center=beam_center,
             scan_info=scan_info,

tests/tests_devices/test_eiger.py

@@ -23,7 +23,12 @@ from ophyd import Staged
 from ophyd_devices.utils.psi_device_base_utils import DeviceStatus
 
 from csaxs_bec.devices.jungfraujoch.eiger import EigerError
-from csaxs_bec.devices.jungfraujoch.eiger import EIGER_READOUT_TIME, MIN_EXP_TIME
+from csaxs_bec.devices.jungfraujoch.eiger import (
+    EIGER_READOUT_TIME,
+    MIN_EXP_TIME,
+    MISSING_PACKETS_MESSAGE,
+    RAISE_ON_MISSING_PACKETS,
+)
 from csaxs_bec.devices.jungfraujoch.eiger_1_5m import EIGER1_5M_READOUT_TIME, Eiger1_5M
 from csaxs_bec.devices.jungfraujoch.eiger_9m import EIGER9M_READOUT_TIME, Eiger9M
 
@@ -329,11 +334,12 @@ def test_eiger_on_complete(eiger_1_5m, raise_timeout):
         assert status.success == False if raise_timeout else True
 
 
-@pytest.mark.skip(
-    reason="This test currently fails because the error should be skipped for now due to HW issues."
-)
 def test_eiger_on_complete_error_message(eiger_1_5m):
-    """Test that on_complete raises if the message severity is error."""
+    """Test that on_complete raises if the message severity is error.
+
+    Unrelated to the missing-packet tolerance: this message is a different error, so it
+    raises whatever raise_on_missing_packets is set to.
+    """
     # Test that on_complete raises if message_severity is error
     eiger = eiger_1_5m
     unblock_wait_for_idle = threading.Event()
@@ -571,3 +577,125 @@ def test_eiger_subclasses_name_readout_time_for_the_device_server():
     """
     for cls in (Eiger1_5M, Eiger9M):
         assert "readout_time" in inspect.signature(cls).parameters
+
+
+def _run_on_complete(eiger, broker_status, images_collected=0, timeout=5):
+    """Drive on_complete to a verdict against a given broker status.
+
+    Returns the status once it resolves; raises whatever on_complete raised.
+    """
+    eiger._wait_for_on_complete = 1  # keep the loop short
+    unblock_wait_for_idle = threading.Event()
+
+    def mock_wait_for_idle(timeout: float, raise_on_timeout: bool) -> bool:
+        return unblock_wait_for_idle.wait(timeout)
+
+    with (
+        mock.patch.object(eiger.jfj_client.api, "status_get", return_value=broker_status),
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", side_effect=mock_wait_for_idle),
+        mock.patch.object(
+            eiger.jfj_client.api,
+            "statistics_data_collection_get",
+            return_value=MeasurementStatistics(run_number=1, images_collected=images_collected),
+        ),
+    ):
+        status = eiger.complete()
+        unblock_wait_for_idle.set()
+        status.wait(timeout=timeout)
+    return status
+
+
+def _missing_packets_status():
+    """A broker status carrying the packet-loss error JungfrauJoch actually emits."""
+    return BrokerStatus(
+        state="Idle", message_severity="error", message=f"{MISSING_PACKETS_MESSAGE} (run 42)"
+    )
+
+
+def test_eiger_tolerates_missing_packets_when_configured_to(eiger_1_5m):
+    """With the tolerance on, a missing-packet error completes the scan and is counted."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+
+    status = _run_on_complete(eiger, _missing_packets_status())
+
+    assert status.done is True
+    assert status.success is True
+    assert eiger.get_missing_packet_events() == 1
+
+
+def test_eiger_raises_on_missing_packets_when_asked_to(eiger_1_5m):
+    """With the tolerance off, the same error fails the scan and is not counted."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(True)
+
+    with pytest.raises(EigerError):
+        _run_on_complete(eiger, _missing_packets_status())
+
+    assert eiger.get_missing_packet_events() == 0
+
+
+def test_eiger_tolerance_does_not_extend_to_other_broker_errors(eiger_1_5m):
+    """Tolerating packet loss must not tolerate every error the broker reports."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+
+    with pytest.raises(EigerError):
+        _run_on_complete(
+            eiger, BrokerStatus(state="Idle", message_severity="error", message="Detector on fire")
+        )
+
+    assert eiger.get_missing_packet_events() == 0
+
+
+def test_eiger_tolerance_does_not_hide_a_short_acquisition(eiger_1_5m):
+    """The frame-count check still fires: only flagged-but-complete data is let through.
+
+    This is what makes the tolerance defensible -- a scan that actually lost frames still
+    fails, whatever raise_on_missing_packets says.
+    """
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+    eiger._num_triggers = 10
+
+    with pytest.raises(EigerError, match="acquisition incomplete"):
+        _run_on_complete(eiger, _missing_packets_status(), images_collected=9)
+
+
+def test_eiger_missing_packet_tolerance_is_switchable_at_runtime(eiger_1_5m):
+    """The flag is reachable from the client, so a beamtime can change its mind."""
+    eiger = eiger_1_5m
+    assert eiger.get_raise_on_missing_packets() is RAISE_ON_MISSING_PACKETS
+
+    eiger.set_raise_on_missing_packets(True)
+    assert eiger.get_raise_on_missing_packets() is True
+    eiger.set_raise_on_missing_packets(False)
+    assert eiger.get_raise_on_missing_packets() is False
+
+    for name in (
+        "get_raise_on_missing_packets",
+        "set_raise_on_missing_packets",
+        "get_missing_packet_events",
+    ):
+        assert name in type(eiger).USER_ACCESS
+
+
+def test_eiger_missing_packet_tolerance_is_overridable_per_deployment(mock_scan_info):
+    """deviceConfig's raise_on_missing_packets wins over the default, per detector."""
+    dev = Eiger1_5M(
+        name="eiger_1_5m",
+        beam_center=(256, 256),
+        detector_distance=100.0,
+        raise_on_missing_packets=True,
+    )
+    try:
+        assert dev.get_raise_on_missing_packets() is True
+    finally:
+        dev.destroy()
+
+
+def test_eiger_subclasses_name_raise_on_missing_packets_for_the_device_server():
+    """Same rule as readout_time: a deviceConfig key only reaches a device if the
+    class signature names it, so **kwargs alone would silently drop it."""
+    for cls in (Eiger1_5M, Eiger9M):
+        assert "raise_on_missing_packets" in inspect.signature(cls).parameters