feat(panda): only arm PandA for flomni_fermat_scan

Generic BEC staging (stage_all_devices/pre_scan_all_devices) has no
per-scan device opt-in -- every enabled device is staged/pre_scanned for
every scan type (grid_scan, line_scan, alignment moves, ...), not just the
ones that consume its data. deviceTags (ptycho_flomni) are never consulted
by that code path, they're purely a client-side grouping convenience. So
omny_panda_continuous was being Arm()ed and disarmed for every scan in a
session even though only flomni_fermat_scan reads its data -- unnecessary
hardware arm/disarm cycling, and it directly compounds the stale
frame_counter carryover bug documented in PANDA_POSITION_VALIDATION.md
sec 7 (more arm cycles means more abort->restage windows where that bug
can be triggered).

PandaBoxOMNY now takes an arm_scan_allowlist constructor kwarg (None by
default, arms for every scan -- unchanged behavior). When set, on_stage()
computes _should_arm_panda from scan_parameters.scan_name against the
list; on_pre_scan() skips sending Arm() entirely when it's False, and
on_complete() skips the wait-for-frame-count poll loop the same way, both
resolving immediately instead. The data-readout thread/TCP connection
startup in on_stage() is untouched, since it doesn't touch PandA's
hardware register state (COUNTER1) the way Arm() does.

ptycho_flomni.yaml's omny_panda_continuous now sets
arm_scan_allowlist: [flomni_fermat_scan].

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/PANDA_POSITION_VALIDATION.md

@@ -269,3 +269,14 @@ reads/writes the same as the buffer fields, and (b) also calling
 `_try_arm`, so an aborted acquisition can't leave stale state for the next
 one to inherit. Re-check against a scan where the stale run's rows show
 `gate_detector_active == 1` before considering this fully closed.
+
+**Mitigation added 2026-09-22:** `omny_panda_continuous` was being armed and
+disarmed for *every* scan (any `grid_scan`/`line_scan`/alignment move, not
+just `flomni_fermat_scan`), since generic BEC staging has no per-scan device
+opt-in and `deviceTags` aren't consulted by it — see "Conditional arming
+(`arm_scan_allowlist`)" in `docs/developer/panda_box_free_running_setup.md`.
+It's now only actually `Arm()`ed for `flomni_fermat_scan` via the new
+`arm_scan_allowlist` device config, which directly cuts down how often this
+class of bug can even be triggered (fewer arm cycles → fewer abort→restage
+windows like `S00388`'s), independent of whatever the remaining root cause
+above turns out to be.

csaxs_bec/device_configs/ptycho_flomni.yaml

@@ -707,6 +707,13 @@ omny_panda_continuous:
   deviceConfig:
     host: omny-panda.psi.ch
     raw_stream_mode: true
+    # Only actually Arm() PandA for scans that consume its data -- generic BEC staging
+    # (stage_all_devices/pre_scan_all_devices) otherwise arms this device for every scan type
+    # (grid_scan, line_scan, alignment moves, ...), not just the ptycho scan that reads it. See
+    # docs/developer/panda_box_free_running_setup.md and
+    # AI_docs/PANDA_POSITION_VALIDATION.md sec 7 (stale frame_counter exposure).
+    arm_scan_allowlist:
+      - flomni_fermat_scan
     signal_alias:
       FMC_IN.VAL1.Value: cap_voltage_fzp_y
       FMC_IN.VAL2.Value: cap_voltage_fzp_x

csaxs_bec/devices/panda_box/panda_box_omny.py

@@ -82,6 +82,7 @@ class PandaBoxOMNY(PandaBox):
         raw_stream_flush_interval: float = 0.05,
         raw_stream_flush_row_count: int = 500,
         pre_scan_timeout: float = 5.0,
+        arm_scan_allowlist: list[str] | None = None,
         **kwargs,
     ) -> None:
         # Free-running raw-stream configuration. Defaults preserve today's behavior:
@@ -99,6 +100,15 @@ class PandaBoxOMNY(PandaBox):
         # since on_pre_scan retries re-arm the box (and thus reset COUNTER1's counting) too.
         self._raw_stream_seen_reset = False
 
+        # Scan names (ScanServerScanInfo.scan_name, e.g. "flomni_fermat_scan") this device
+        # should actually be armed for. None (default) means "arm for every scan" -- today's
+        # behavior, fully backward compatible. Generic BEC staging (stage_all_devices/
+        # pre_scan_all_devices in bec_server) arms every enabled device for every scan type,
+        # not just the ones that consume this device's data; this lets a beamline config opt
+        # this device out of Arm()/COUNTER1 cycling (and the associated stale-frame_counter
+        # exposure, see AI_docs/PANDA_POSITION_VALIDATION.md sec 7) for scans that don't need it.
+        self.arm_scan_allowlist = arm_scan_allowlist
+
         # Seconds to wait for the PandA READY event per arming attempt in on_pre_scan.
         self.pre_scan_timeout = pre_scan_timeout
         # Set while the data readout thread is parked (not inside _run_data_readout), so a
@@ -120,11 +130,28 @@ class PandaBoxOMNY(PandaBox):
         self._acquisition_group = "burst"
         self._timeout_on_completed = 10
         self.scan_parameters: ScanServerScanInfo | None = None
+        # Safe default (arm): matches arm_scan_allowlist=None semantics, and covers the case
+        # where on_pre_scan/on_complete are called before any on_stage (some tests do this).
+        self._should_arm_panda: bool = True
+
+    def _compute_should_arm_panda(self) -> bool:
+        """Whether this scan should actually Arm() the PandA and wait on completion. See the
+        class-level note on arm_scan_allowlist. Fails open (arms) whenever the allowlist isn't
+        set or scan_parameters isn't available yet, preserving today's unconditional-arm
+        behavior in both cases."""
+        if self.arm_scan_allowlist is None:
+            return True
+        if self.scan_parameters is None:
+            return True
+        return self.scan_parameters.scan_name in self.arm_scan_allowlist
 
     def on_stage(self):
         start_time = time.time()
         super().on_stage()
         self.scan_parameters = fetch_scan_info(self.scan_info)
+        # Recomputed fresh every stage cycle: the same device instance is staged for many
+        # different scan types across a session, not just the ones in arm_scan_allowlist.
+        self._should_arm_panda = self._compute_should_arm_panda()
 
         self._reset_raw_stream_state()
 
@@ -156,7 +183,15 @@ class PandaBoxOMNY(PandaBox):
         """Arm the PCAP module. Unlike the base class this does not wait forever for the READY
         event: if it does not arrive, the box is forcibly reset and arming is retried once,
         and only if that fails as well the returned status fails, so the scan is aborted with
-        an error instead of hanging silently in pre_scan."""
+        an error instead of hanging silently in pre_scan.
+
+        If arm_scan_allowlist excludes the current scan (see _should_arm_panda/on_stage), PandA
+        is left untouched entirely -- no Arm() is sent -- and this resolves immediately.
+        """
+        if not self._should_arm_panda:
+            status = StatusBase(obj=self)
+            status.set_finished()
+            return status
         # Not registered with cancel_on_stop: the status of each single attempt in _try_arm is,
         # so a stop makes the task raise DeviceStoppedError, which fails this status. Cancelling
         # this one as well would make the task handler try to resolve it a second time.
@@ -237,7 +272,17 @@ class PandaBoxOMNY(PandaBox):
         return super().on_unstage()
 
     def on_complete(self):
-        """On complete is called after the scan is complete. We need to wait for the capture to complete before we can disarm the PandaBox."""
+        """On complete is called after the scan is complete. We need to wait for the capture to
+        complete before we can disarm the PandaBox -- unless this device was never armed for
+        this scan (see _should_arm_panda/on_stage), in which case there is nothing to wait for
+        and no wait-loop is started. _disarm() is deliberately not called here in that case:
+        the box is already guaranteed disarmed (on_stage waits for DISARMED before staging),
+        nothing armed it since Arm() was skipped, and on_unstage()/stop() unconditionally
+        disarm anyway at the end of the lifecycle."""
+        if not self._should_arm_panda:
+            status = StatusBase(obj=self)
+            status.set_finished()
+            return status
         if self.raw_stream_mode:
             return self._on_complete_free_running()
         return self._on_complete_burst()

docs/developer/panda_box_free_running_setup.md

@@ -284,6 +284,35 @@ verification" row needs one caveat: a single backward step at the very start
 of the raw sequence, from the previous acquisition's terminal value down to
 `0`, is this expected (now-filtered) artifact, not a sign of dropped rows.
 
+## Conditional arming (`arm_scan_allowlist`, added 2026-09-22)
+
+Generic BEC staging (`stage_all_devices`/`pre_scan_all_devices` in
+`bec_server.scan_server.scans.scan_actions`) has no concept of "this device
+only matters for scan X" -- every `enabled: true` device gets `stage()`/
+`pre_scan()`/`unstage()` called for *every* scan that runs (`grid_scan`,
+`line_scan`, alignment moves, ...), not just the ones that actually consume
+its data. `deviceTags` (e.g. `ptycho_flomni`) are never consulted by that
+staging code -- they're purely a client-side grouping convenience. So
+without any further gating, `omny_panda_continuous` was being `Arm()`ed and
+disarmed for every single scan run during a session, even though only
+`flomni_fermat_scan` reads its data. That's unnecessary hardware arm/disarm
+cycling, and it directly compounds the stale-`frame_counter` bug above: more
+arm cycles mean more windows where a mid-scan abort can leave a stale
+counter value for the *next* acquisition to inherit (see
+`AI_docs/PANDA_POSITION_VALIDATION.md` sec 7 for a real incident traced to
+exactly this).
+
+`PandaBoxOMNY` now takes an `arm_scan_allowlist: list[str] | None` constructor
+kwarg (`None` by default -- arm for every scan, unchanged behavior). When
+set, `on_stage()` computes `_should_arm_panda` from
+`scan_parameters.scan_name` against that list; `on_pre_scan()` skips sending
+`Arm()` entirely (resolves immediately) when it's `False`, and `on_complete()`
+skips the wait-for-frame-count poll loop the same way. `on_stage()`'s own
+readout-thread/TCP-connection startup is untouched by this -- it doesn't
+touch PandA's hardware register state (`COUNTER1`), unlike `Arm()`, so it
+stays out of scope. `ptycho_flomni.yaml`'s `omny_panda_continuous` sets
+`arm_scan_allowlist: [flomni_fermat_scan]`.
+
 ## Switching between modes
 
 `omny_panda` (Mode A) and `omny_panda_continuous` (Mode B) need different PandA hardware

tests/tests_devices/test_panda.py

@@ -484,6 +484,135 @@ def test_panda_omny_pre_scan_arm_resets_raw_stream_state(panda_omny):
         mock_reset_raw.assert_called_once()
 
 
+def test_panda_omny_arm_scan_allowlist_defaults_to_none(panda_omny):
+    """A device constructed without arm_scan_allowlist must arm for every scan (today's
+    behavior), both via the raw config attribute and the computed should-arm flag."""
+    dev = panda_omny
+    assert dev.arm_scan_allowlist is None
+    assert dev._should_arm_panda is True
+
+
+def test_panda_omny_stage_should_arm_stays_true_when_allowlist_none(panda_omny):
+    """Backward-compatibility regression: staging with the default allowlist (None) must keep
+    _should_arm_panda True regardless of the scan_name in scan_info."""
+    dev = panda_omny
+    dev.stage()
+    assert dev._should_arm_panda is True
+
+
+def test_panda_omny_stage_should_arm_true_when_scan_name_in_allowlist(panda_omny):
+    dev = panda_omny
+    dev.arm_scan_allowlist = ["flomni_fermat_scan"]
+    dev.scan_info.msg.info["scan_name"] = "flomni_fermat_scan"
+    dev.stage()
+    assert dev._should_arm_panda is True
+
+
+def test_panda_omny_stage_should_arm_false_when_scan_name_not_in_allowlist(panda_omny):
+    dev = panda_omny
+    dev.arm_scan_allowlist = ["flomni_fermat_scan"]
+    dev.scan_info.msg.info["scan_name"] = "grid_scan"
+    dev.stage()
+    assert dev._should_arm_panda is False
+
+
+def test_panda_omny_should_arm_recomputed_every_stage(panda_omny):
+    """The same device instance is staged for many different scan types across a session; the
+    should-arm decision must be recomputed fresh on every stage, not stick from a previous
+    scan."""
+    dev = panda_omny
+    dev.arm_scan_allowlist = ["flomni_fermat_scan"]
+
+    with mock.patch.object(dev, "_disarm"):
+        dev.scan_info.msg.info["scan_name"] = "grid_scan"
+        dev.stage()
+        assert dev._should_arm_panda is False
+        dev.unstage()
+
+        dev.scan_info.msg.info["scan_name"] = "flomni_fermat_scan"
+        dev.stage()
+        assert dev._should_arm_panda is True
+
+
+def test_panda_omny_pre_scan_skips_arm_when_should_arm_false(panda_omny):
+    """When arm_scan_allowlist excludes the current scan, on_pre_scan must resolve immediately
+    without sending Arm() or spinning up the retry task at all."""
+    dev = panda_omny
+    dev._should_arm_panda = False
+    with (
+        mock.patch.object(dev, "_arm") as mock_arm,
+        mock.patch.object(dev, "_arm_with_retry") as mock_arm_with_retry,
+        mock.patch.object(dev.task_handler, "submit_task") as mock_submit_task,
+    ):
+        status = dev.on_pre_scan()
+        assert status.done is True
+        assert status.success is True
+        mock_arm.assert_not_called()
+        mock_arm_with_retry.assert_not_called()
+        mock_submit_task.assert_not_called()
+
+
+def test_panda_omny_pre_scan_arms_when_should_arm_true_via_allowlist(panda_omny):
+    """End-to-end: a scan_name matching arm_scan_allowlist still goes through the full arm/
+    retry machinery exactly as with the default (None) allowlist."""
+    dev = panda_omny
+    dev.pre_scan_timeout = 2
+    dev.arm_scan_allowlist = ["flomni_fermat_scan"]
+    dev.scan_info.msg.info["scan_name"] = "flomni_fermat_scan"
+    dev.stage()
+    assert dev._should_arm_panda is True
+    with (
+        mock.patch.object(dev, "_arm") as mock_arm,
+        mock.patch.object(dev, "_reset_panda") as mock_reset,
+    ):
+        _deliver_ready_after(dev, 0.05)
+        status = dev.on_pre_scan()
+        status.wait(timeout=4)
+        assert status.success is True
+        mock_arm.assert_called_once()
+        mock_reset.assert_not_called()
+
+
+def test_panda_omny_complete_skips_wait_loop_when_should_arm_false(panda_omny):
+    """When this device was never armed for the current scan, on_complete must resolve
+    immediately without polling PandA or disarming it."""
+    dev = panda_omny
+    dev._should_arm_panda = False
+    with (
+        mock.patch.object(dev, "send_raw") as mock_send_raw,
+        mock.patch.object(dev, "_disarm") as mock_disarm,
+        mock.patch.object(dev.task_handler, "submit_task") as mock_submit_task,
+    ):
+        status = dev.on_complete()
+        assert status.done is True
+        assert status.success is True
+        mock_send_raw.assert_not_called()
+        mock_disarm.assert_not_called()
+        mock_submit_task.assert_not_called()
+
+
+def test_panda_omny_complete_skips_wait_loop_when_should_arm_false_raw_stream_mode(
+    panda_omny_raw_stream,
+):
+    """Same as test_panda_omny_complete_skips_wait_loop_when_should_arm_false, but on the
+    raw_stream_mode=True path that would otherwise dispatch to _on_complete_free_running."""
+    dev = panda_omny_raw_stream
+    dev._should_arm_panda = False
+    with (
+        mock.patch.object(dev, "send_raw") as mock_send_raw,
+        mock.patch.object(dev, "_disarm") as mock_disarm,
+        mock.patch.object(dev, "_flush_raw_stream_buffer") as mock_flush,
+        mock.patch.object(dev.task_handler, "submit_task") as mock_submit_task,
+    ):
+        status = dev.on_complete()
+        assert status.done is True
+        assert status.success is True
+        mock_send_raw.assert_not_called()
+        mock_disarm.assert_not_called()
+        mock_flush.assert_not_called()
+        mock_submit_task.assert_not_called()
+
+
 def test_panda_omny_raw_stream_signal_alias(panda_omny_raw_stream):
     all_signal_names = [name for name, _ in panda_omny_raw_stream.data.signals]
     assert "gate_detector_active" in all_signal_names