fix(flomni): stop duplicate 360deg tomo angles, add sample owner, fix abort crash

Three independent fixes:

- sub_tomo_scan(): in 360-degree tomo_type-1 mode, every sub-tomogram used
  to sweep the full 0-360 range, so each one's own fine grid contained
  angle pairs exactly 180deg apart -- redundant tomographic information.
  Sub-tomograms now each cover a 180deg span, split into low/high halves
  by subtomo_number % 4 with a bit-reversal-of-4 phase table, so adjacent
  pairs (1,2)/(3,4)/(5,6)/(7,8), either quartet, and all 8 combined each
  independently form a complete, evenly-spaced 360deg tomogram at
  successively finer spacing. Total projection count for a given
  tomo_angle_stepsize is now identical between 180 and 360 mode (same N,
  no longer doubled). Updated the 4 other consumers of the old N/step
  formula (zero-deg reference gating, _tomo_type1_actual_grid, the
  parameter wizard, the PDF report) to match. 180-degree mode is
  unchanged. Live-verified against the running flomni sim: real motor
  motion traces the expected boustrophedon path with no duplicate or
  180deg-apart angles.

- Sample storage: added an owner field, packed into the same EPICS DESC
  field as the sample name ("name | owner", via new sample_desc_codec.py)
  since there's no separate PV for it. Wired through
  FlomniSampleStorage, the CLI (flomni_modify_storage_non_interactive,
  ftransfer_modify_storage), the two transfer routines that forward a
  raw DESC value across a gripper move (now unpacked/repacked so owner
  survives the move instead of being dropped or double-packed), and
  SampleStorageWidget. Scoped to flomni only this session; OMNY's
  storage/transfer mixin is unchanged.

- ConsoleButtonsWidget's ABORT button used to send SIGINT then, 500ms
  later, a stop_devices() broadcast to ALL devices with no stop_id --
  an un-suppressed error from that broadcast landing on a queue-tracked
  instruction could kill the scan worker thread outright, requiring a
  full BEC restart. Replaced with: queue.request_scan_abortion() (safe
  no-op if idle, but registers a stop_id so expected errors are
  suppressed), then SIGINT, then a direct, immediate Galil hard stop via
  new GalilController.hard_abort_and_restore_positioning_mode() -- the
  same method ftransfer_abort() now delegates to, so the CLI and GUI
  paths can't drift apart again. SIGINT is sent before the hard stop:
  live testing showed that if the hard stop's mntprgs-clearing side
  effect lands first, a same-session polling loop can mistake it for
  normal completion and fall through into ensure_gripper_up(), which
  must not happen mid-transfer -- sending SIGINT first (near-instant)
  gives that loop's own KeyboardInterrupt handler a head start before
  the hard stop's own multi-step sequence completes. The button is
  labeled per beamline (e.g. "Flomni Motion Stop") and disabled rather
  than silently inert when no hard-stop device is configured/enabled.
  Scoped to flomni only this session (OMNY/LamNI wiring deferred).
  Live-verified against the running sim, including the exact
  stop-lands-mid-queued-instruction scenario that previously crashed
  the scan worker.

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

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -27,6 +27,8 @@ from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import (
     PtychoReconstructor,
     TomoIDManager,
 )
+from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
+from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
 
 # from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
 #        FlomniWebpageGenerator,
@@ -789,7 +791,8 @@ class FlomniSampleTransferMixin:
         self.ensure_gripper_up()
 
         signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-        self.flomni_modify_storage_non_interactive(100, 1, signal_name.get())
+        name, owner = unpack_desc(signal_name.get())
+        self.flomni_modify_storage_non_interactive(100, 1, name, owner=owner)
         self.flomni_modify_storage_non_interactive(position, 0, "-")
 
     def ftransfer_show_all(self):
@@ -846,9 +849,9 @@ class FlomniSampleTransferMixin:
         self.ftransfer_controller_disable_mount_mode()
         self.ensure_gripper_up()
 
-        sample_name = dev.flomni_samples.sample_in_gripper_name.get()
+        sample_name, sample_owner = unpack_desc(dev.flomni_samples.sample_in_gripper_name.get())
         self.flomni_modify_storage_non_interactive(100, 0, "-")
-        self.flomni_modify_storage_non_interactive(position, 1, sample_name)
+        self.flomni_modify_storage_non_interactive(position, 1, sample_name, owner=sample_owner)
 
         if position == 0:
             self.ftransfer_flomni_stage_in()
@@ -859,7 +862,7 @@ class FlomniSampleTransferMixin:
         Get the name of the sample currently in the given position.
         """
         signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-        return signal_name.get()
+        return unpack_desc(signal_name.get())[0]
 
     def ftransfer_sample_change(self, new_sample_position: int):
         self.check_tray_in()
@@ -971,19 +974,24 @@ class FlomniSampleTransferMixin:
     def ftransfer_modify_storage(self, position: int, used: int):
         if used:
             name = input("What's the name of this sample? ")
+            owner = input("Sample owner (optional): ")
         else:
             name = "-"
-        self.flomni_modify_storage_non_interactive(position, used, name)
+            owner = ""
+        self.flomni_modify_storage_non_interactive(position, used, name, owner=owner)
 
-    def flomni_modify_storage_non_interactive(self, position: int, used: int, name: str):
+    def flomni_modify_storage_non_interactive(
+        self, position: int, used: int, name: str, owner: str = ""
+    ):
+        packed_name = pack_desc(name, owner)
         if position == 100:
             dev.flomni_samples.sample_in_gripper.set(used)
-            dev.flomni_samples.sample_in_gripper_name.set(name)
+            dev.flomni_samples.sample_in_gripper_name.set(packed_name)
         else:
             signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
             signal.set(used)
             signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
-            signal_name.set(name)
+            signal_name.set(packed_name)
 
     def check_position_is_valid(self, position: int):
         if 0 <= position < 21:
@@ -1014,33 +1022,21 @@ class FlomniSampleTransferMixin:
         """
         Hard abort of a running sample transfer routine on the Galil controller.
 
-        Stops the controller via dev.ftransy.stop(), which publishes a stop
+        Delegates to GalilController.hard_abort_and_restore_positioning_mode()
-        request that the device server turns into motor.stop() and thus
+        -- the SAME method the GUI's hard-stop button calls directly (see
-        XQ#STOP on the controller. #STOP halts the transfer thread (3),
+        ConsoleButtonsWidget._on_abort()) -- so there is exactly one
-        aborts all motion (AB1) and clears mntprgs/mntmod. Afterwards the
+        implementation of this safety-critical stop sequence shared by both
-        controller is put back into positioning mode.
+        the CLI (Ctrl+C) and GUI paths.
 
         Deliberately does NOT call ensure_gripper_up(): after a mid-transfer
         abort the gripper may be closed around a partially inserted sample,
         so any recovery motion must be assessed and performed manually.
         """
         print("Aborting sample transfer: stopping the controller routine.")
-        dev.ftransy.stop()
+        try:
-        # The stop request is asynchronous (Redis -> device server). Wait
+            dev.ftransy.controller.hard_abort_and_restore_positioning_mode()
-        # until the transfer thread is actually halted before switching mode:
+        except GalilError as exc:
-        # #POSMODE refuses while mntprgs=1 and disable_mount_mode would raise.
+            raise FlomniError(str(exc)) from exc
-        timeout = 5
-        start = time.time()
-        while dev.ftransy.controller.is_thread_active(3):
-            if time.time() - start > timeout:
-                raise FlomniError(
-                    "Transfer abort requested but the controller transfer routine (thread 3)"
-                    f" did not stop within {timeout} s. Check the controller."
-                )
-            time.sleep(0.1)
-        # Ensure the controller is back in positioning mode. #STOP already
-        # clears mntmod, so this is mostly a verification step.
-        self.ftransfer_controller_disable_mount_mode()
 
     def ftransfer_confirm(self, step_name: str = ""):
         confirm = int(float(dev.ftransy.controller.socket_put_and_receive("MG confirm").strip()))
@@ -2305,6 +2301,121 @@ class Flomni(
             f.write(" ".join(map(str, x_vals)) + "\n")
             f.write(" ".join(map(str, zeros)) + "\n")
 
+    @staticmethod
+    def _subtomo_angle_plan(subtomo_number, tomo_angle_range, tomo_angle_stepsize, start_angle=None):
+        """
+        Pure angle-generation logic for one sub-tomogram of an equally
+        spaced 8-sub-tomogram tomography scan (tomo_type == 1). No device
+        I/O / progress side effects - kept separate from sub_tomo_scan()
+        so the angle math is directly unit-testable.
+
+        N/step are always computed against a fixed 180 degree span,
+        independent of tomo_angle_range: total projection count for a
+        given tomo_angle_stepsize is identical whether tomo_angle_range
+        is 180 or 360. In 180 mode all 8 sub-tomograms cover the same
+        [0,180) span (bit-reversal interlacing at step/8, unchanged from
+        the original scheme). In 360 mode each sub-tomogram instead
+        covers only a 180-degree span - never the full circle - split by
+        subtomo_number % 4 into a "low" half [0,180) (n%4 in (1,0)) and a
+        "high" half [180,360) (n%4 in (2,3)), with the 4 sub-tomograms
+        serving each half interlaced via a bit-reversal of {0,1,2,3}
+        (not sequential order): this specific permutation is what makes
+        adjacent pairs (1,2)/(3,4)/(5,6)/(7,8), either quartet (1-4 or
+        5-8), and all 8 combined each independently form a complete,
+        evenly-spaced 360-degree tomogram at successively finer spacing
+        (step, step/2, step/4) - a sequential assignment would leave gaps
+        in the quartet-level grids. Without this split, every
+        sub-tomogram's own 360-degree sweep would contain angle pairs
+        exactly 180 degrees apart, which is redundant tomographic
+        information.
+
+        Returns:
+            angles (np.ndarray): the N angles (degrees) for this sub-tomogram.
+            subtomo_offset (int): index of this sub-tomogram's first angle
+                within the combined progress numbering (0 unless resuming).
+            N (int): number of projections in this (and every) sub-tomogram.
+            step (float): degree spacing between consecutive points within
+                this sub-tomogram.
+        """
+        explicit_start_angle = start_angle is not None
+
+        # N/step not guaranteed to be a whole/exact division of the
+        # configured tomo_angle_stepsize; N is the actual, integer number
+        # of projections per sub-tomogram, step is the step size that's
+        # ACTUALLY achievable while landing exactly on N evenly-spaced
+        # points across a 180 degree span. This corrected step - not the
+        # raw, configured tomo_angle_stepsize - is used for BOTH the
+        # per-point ramp AND the inter-sub-tomogram phase offsets below.
+        N = int(180.0 / tomo_angle_stepsize)
+        step = 180.0 / N
+
+        if tomo_angle_range == 180:
+            base = 0.0
+            phase_eighths = {1: 0, 2: 4, 3: 2, 4: 6, 5: 1, 6: 5, 7: 3, 8: 7}
+            phase = step / 8.0 * phase_eighths[subtomo_number]
+            forward = bool(subtomo_number % 2)
+        else:
+            quarter = {1: 0, 4: 2, 5: 1, 8: 3, 2: 0, 3: 2, 6: 1, 7: 3}
+            mod4 = subtomo_number % 4
+            base = 0.0 if mod4 in (1, 0) else 180.0
+            phase = step / 4.0 * quarter[subtomo_number]
+            forward = mod4 in (1, 2)
+
+        if not explicit_start_angle:
+            if forward:
+                # Low end of this sub-tomogram's angular phase (same
+                # convention regardless of direction - it's what makes the
+                # combined sub-tomograms interlace into one fine angular
+                # grid).
+                start_angle = base + phase
+            else:
+                # A reverse sweep must begin at the HIGH end of this
+                # sub-tomogram's 180-degree span and descend. The literal
+                # high end (base + 180) would itself be rejected by
+                # _tomo_scan_at_angle's own "angle < tomo_angle_range +
+                # 0.05" gate for every sub-tomogram whose phase is nonzero
+                # (it lands just past the range), so start one step below
+                # that instead - this is the angle that will actually be
+                # the first one accepted. Skipped when start_angle is given
+                # explicitly (i.e. resuming mid sub-tomogram), since then
+                # the value is already the literal current angle.
+                start_angle = base + 180.0 - step + phase
+
+        # Every sub-tomogram covers exactly N projections, matching
+        # subtomo_total_projections elsewhere in this class - generated by
+        # plain arithmetic at the exact (corrected) step size, with no
+        # clamping. This deliberately never generates the boundary point at
+        # start +/- 180: that point is silently rejected by
+        # _tomo_scan_at_angle's own range gate for every sub-tomogram whose
+        # phase is nonzero anyway, so generating it only ever produced an
+        # inconsistent extra projection for the phase==0 sub-tomogram while
+        # every other sub-tomogram was silently one projection short.
+        if forward:
+            angles = start_angle + np.arange(N) * step
+        else:
+            angles = start_angle - np.arange(N) * step
+
+        if not explicit_start_angle:
+            # normal operation: always start at zero
+            subtomo_offset = 0
+        else:
+            # Explicitly subtract base and phase before dividing, rather
+            # than relying on an algebraic shortcut: for some sub-tomograms
+            # (e.g. #2 in 180 mode, #3/#4 in 360 mode), phase/step is
+            # exactly 0.5, landing precisely on the float rounding
+            # tie-break boundary - floating-point noise there
+            # unpredictably rounds up or down, which previously gave the
+            # wrong offset (off by +1) in ~44% of resumes within subtomo 2
+            # specifically (verified by exhaustive sweep). Every other
+            # sub-tomogram's phase fraction is far enough from 0.5 that an
+            # algebraic shortcut never broke for them.
+            if forward:
+                subtomo_offset = round((start_angle - base - phase) / step)
+            else:
+                subtomo_offset = round(((base + 180.0 - step + phase) - start_angle) / step)
+
+        return angles, subtomo_offset, N, step
+
     def sub_tomo_scan(self, subtomo_number, start_angle=None):
         """
         Performs a sub tomogram scan.
@@ -2313,117 +2424,21 @@ class Flomni(
             start_angle (float, optional): The start angle of the scan. Defaults to None.
         """
 
-        explicit_start_angle = start_angle is not None
+        if start_angle is not None:
-        if explicit_start_angle:
             print(f"Sub tomo scan with start angle {start_angle} requested.")
 
-        # tomo_angle_range / tomo_angle_stepsize is not guaranteed to be a
+        angles, subtomo_offset, N, step = self._subtomo_angle_plan(
-        # whole number (e.g. a "total number of projections" that isn't a
+            subtomo_number, self.tomo_angle_range, self.tomo_angle_stepsize, start_angle=start_angle
-        # multiple of 8 was configured). N is the actual, integer number of
+        )
-        # projections per sub-tomogram; step is the step size that's
-        # ACTUALLY achievable while landing exactly on N evenly-spaced
-        # points across tomo_angle_range. This corrected step - not the
-        # raw, configured tomo_angle_stepsize - is used for BOTH the
-        # per-point ramp AND the inter-sub-tomogram phase offsets below.
-        # Using the raw stepsize for the phase offsets while the ramp used
-        # the corrected one is what caused the combined/interlaced
-        # tomogram to have an inconsistent angular spacing whenever N
-        # wasn't already a whole number for the raw stepsize.
-        N = int(self.tomo_angle_range / self.tomo_angle_stepsize)
-        step = self.tomo_angle_range / N
-
-        # Phase offset (degrees) for this sub-tomogram's position in the
-        # bit-reversal interlacing order - needed below to correctly
-        # recover the loop index i when resuming with an explicit
-        # start_angle (see the i==0 block further down).
-        phase_eighths = {1: 0, 2: 4, 3: 2, 4: 6, 5: 1, 6: 5, 7: 3, 8: 7}
-        phase = step / 8.0 * phase_eighths[subtomo_number]
-
-        if start_angle is None:
-            if subtomo_number == 1:
-                start_angle = 0
-            elif subtomo_number == 2:
-                start_angle = step / 8.0 * 4
-            elif subtomo_number == 3:
-                start_angle = step / 8.0 * 2
-            elif subtomo_number == 4:
-                start_angle = step / 8.0 * 6
-            elif subtomo_number == 5:
-                start_angle = step / 8.0 * 1
-            elif subtomo_number == 6:
-                start_angle = step / 8.0 * 5
-            elif subtomo_number == 7:
-                start_angle = step / 8.0 * 3
-            elif subtomo_number == 8:
-                start_angle = step / 8.0 * 7
-
-            if not subtomo_number % 2:  # even = reverse
-                # The table above gives the LOW end of this sub-tomogram's
-                # angular phase (same convention as the forward/odd
-                # sub-tomograms - it's what makes the combined 8 sub-tomograms
-                # interlace into one fine angular grid). A reverse sweep must
-                # begin at the HIGH end of that span and descend. The literal
-                # high end (phase + tomo_angle_range) would itself be rejected
-                # by _tomo_scan_at_angle's own "angle < tomo_angle_range + 0.05"
-                # gate for every sub-tomogram whose phase is nonzero (it lands
-                # just past the range), so start one step below that instead -
-                # this is the angle that will actually be the first one
-                # accepted. This step is skipped when start_angle is given
-                # explicitly (i.e. we are resuming mid sub-tomogram), since
-                # then the value is already the literal current angle.
-                start_angle = start_angle + self.tomo_angle_range - step
-
-        # _tomo_shift_angles (potential global variable)
-        _tomo_shift_angles = 0
-        # compute number of projections
-
-        start = start_angle + _tomo_shift_angles
-
-        # Every sub-tomogram covers exactly N projections, matching
-        # subtomo_total_projections elsewhere in this class - generated by
-        # plain arithmetic at the exact (corrected) step size, with no
-        # clamping. This deliberately never generates the boundary point at
-        # start +/- tomo_angle_range: that point is silently rejected by
-        # _tomo_scan_at_angle's own range gate for every sub-tomogram whose
-        # phase is nonzero anyway, so generating it only ever produced an
-        # inconsistent extra projection for the phase==0 sub-tomogram while
-        # every other sub-tomogram was silently one projection short.
-        if subtomo_number % 2:  # odd = forward: low -> high
-            angles = start + np.arange(N) * step
-        else:  # even = reverse: high -> low
-            angles = start - np.arange(N) * step
 
         for i, angle in enumerate(angles):
 
             self.progress["subtomo"] = subtomo_number
 
-            # --- NEW LOGIC FOR OFFSET WHEN start_angle IS SPECIFIED ---
-            if i == 0:
-                if not explicit_start_angle:
-                    # normal operation: always start at zero
-                    self._subtomo_offset = 0
-
-                else:
-                    # Explicitly subtract the phase before dividing, rather
-                    # than relying on an algebraic shortcut: for subtomo 2,
-                    # phase/step is exactly 0.5 (its phase_eighths value is
-                    # 4), landing precisely on the float rounding tie-break
-                    # boundary - floating-point noise there unpredictably
-                    # rounds up or down, which previously gave the wrong
-                    # offset (off by +1) in ~44% of resumes within subtomo 2
-                    # specifically (verified by exhaustive sweep). Every
-                    # other sub-tomogram's phase fraction is far enough from
-                    # 0.5 that the old shortcut never broke for them.
-                    if subtomo_number % 2:  # odd = forward direction
-                        self._subtomo_offset = round((start_angle - phase) / step)
-                    else:  # even = reverse direction
-                        self._subtomo_offset = round(
-                            ((phase + self.tomo_angle_range - step) - start_angle) / step
-                        )
-
             # progress index must always increase
+            if i == 0:
+                self._subtomo_offset = subtomo_offset
             self.progress["subtomo_projection"] = self._subtomo_offset + i
-            # ------------------------------------------------------------
 
             # existing progress fields. N is already an int (by
             # construction, see above), so total_projections = N * 8 is
@@ -2439,6 +2454,16 @@ class Flomni(
             # finally do the scan at this angle
             self._tomo_scan_at_angle(angle, subtomo_number)
 
+    def _subtomo_starts_near_zero(self, subtomo_number: int) -> bool:
+        """True if this sub-tomogram's own natural sweep begins near angle 0
+        (i.e. its direction is forward and its half is the low [0,180) one).
+        Used to gate the zero_deg_reference_at_each_subtomo damage-tracking
+        shot: forcing that shot before a sub-tomogram that doesn't actually
+        start near 0 would mean a large, wasted detour."""
+        if self.tomo_angle_range == 180:
+            return bool(subtomo_number % 2)
+        return subtomo_number % 4 == 1
+
     @staticmethod
     def _retry_unless_flomni_error(exc: Exception, attempt: int) -> bool:
         """scan_repeat() exc_handler: retry any exception except FlomniError.
@@ -2632,14 +2657,23 @@ class Flomni(
                 # 8 equally spaced sub-tomograms
                 self.progress["tomo_type"] = "Equally spaced sub-tomograms"
                 for ii in range(subtomo_start, 9):
-                    if start_angle is None and ii % 2 and self.zero_deg_reference_at_each_subtomo:
+                    if (
+                        start_angle is None
+                        and self._subtomo_starts_near_zero(ii)
+                        and self.zero_deg_reference_at_each_subtomo
+                    ):
                         # Dedicated reference shot at exactly 0 degrees, taken
-                        # every time the rotation passes back through 0 (i.e.
+                        # every time the rotation passes back through 0, for
-                        # at the start of every odd/forward sub-tomogram), for
                         # tracking radiation damage over the full tomogram.
-                        # Skipped when resuming mid-sub-tomogram (start_angle
+                        # In 180 mode that's the start of every odd/forward
-                        # given explicitly) since we're not actually passing
+                        # sub-tomogram; in 360 mode only sub-tomograms 1 and 5
-                        # through 0 deg at that moment.
+                        # (the "low half, forward" group) actually start near
+                        # 0 - sub-tomograms 2/3/6/7 operate entirely in the
+                        # [180,360) half and forcing a detour to 0 before them
+                        # would defeat the point of the 360-mode boustrophedon
+                        # path. Skipped when resuming mid-sub-tomogram
+                        # (start_angle given explicitly) since we're not
+                        # actually passing through 0 deg at that moment.
                         self._tomo_scan_at_angle(0, ii)
                     self.sub_tomo_scan(ii, start_angle=start_angle)
                     start_angle = None
@@ -3537,11 +3571,16 @@ class Flomni(
     def _tomo_type1_actual_grid(self) -> tuple[int, float, int]:
         """Compute the actual (achievable) tomo_type==1 grid from the
         currently stored self.tomo_angle_stepsize -- the SAME way
-        sub_tomo_scan() does it. Returns (N, step, total_projections):
+        sub_tomo_scan()/_subtomo_angle_plan() does it. Returns (N, step,
+        total_projections):
           N: integer number of projections per sub-tomogram
-          step: the achievable per-projection angular step (range / N) --
+          step: the achievable per-projection angular step within a
-                this is what the scan actually runs at, NOT
+                sub-tomogram's own 180-degree span -- this is what the scan
-                self.tomo_angle_stepsize itself
+                actually runs at, NOT self.tomo_angle_stepsize itself. N/step
+                are always computed against a fixed 180 degrees, independent
+                of tomo_angle_range: total projection count for a given
+                tomo_angle_stepsize is the same whether tomo_angle_range is
+                180 or 360 (see _subtomo_angle_plan()'s docstring).
           total_projections: N * 8
 
         self.tomo_angle_stepsize is stored as a raw, uncorrected value
@@ -3554,8 +3593,8 @@ class Flomni(
         the displayed "angular step within sub-tomogram" to silently differ
         from the angle the scan was actually acquiring at.
         """
-        N = int(self.tomo_angle_range / self.tomo_angle_stepsize)
+        N = int(180.0 / self.tomo_angle_stepsize)
-        step = self.tomo_angle_range / N
+        step = 180.0 / N
         return N, step, N * 8
 
     def tomo_parameters(self):
@@ -3695,7 +3734,10 @@ class Flomni(
                 tomo_numberofprojections = self._get_val(
                     "Total number of projections", current_total, int
                 )
-                self.tomo_angle_stepsize = (self.tomo_angle_range / tomo_numberofprojections) * 8
+                # N/step (and therefore total projections) are always
+                # computed against a fixed 180 degrees, independent of
+                # tomo_angle_range -- see _subtomo_angle_plan()'s docstring.
+                self.tomo_angle_stepsize = (180.0 / tomo_numberofprojections) * 8
 
                 # Now report what was ACTUALLY achieved, via the same helper
                 # sub_tomo_scan() effectively uses -- not the raw value just
@@ -4337,15 +4379,19 @@ class Flomni(
         stitching = f"{self.stitch_x:.0f}/{self.stitch_y:.0f}"
         dataset_id = str(self.client.queue.next_dataset_number)
         account = bec.active_account
+        # Same grid sub_tomo_scan() actually uses -- see
+        # _tomo_type1_actual_grid()'s docstring for why this can't just
+        # recompute int((tomo_angle_range/tomo_angle_stepsize)*8) locally.
+        _, _, tomo_type1_total_projections = self._tomo_type1_actual_grid()
         content = [
             f"{'Sample Name:':<{padding}}{self.sample_name:>{padding}}\n",
             f"{'Measurement ID:':<{padding}}{str(self.tomo_id):>{padding}}\n",
             f"{'Dataset ID:':<{padding}}{dataset_id:>{padding}}\n",
             f"{'Sample Info:':<{padding}}{'Sample Info':>{padding}}\n",
             f"{'e-account:':<{padding}}{str(account):>{padding}}\n",
-            f"{'Number of projections:':<{padding}}{int((self.tomo_angle_range / self.tomo_angle_stepsize) * 8):>{padding}}\n",
+            f"{'Number of projections:':<{padding}}{tomo_type1_total_projections:>{padding}}\n",
             f"{'First scan number:':<{padding}}{self.client.queue.next_scan_number:>{padding}}\n",
-            f"{'Last scan number approx.:':<{padding}}{self.client.queue.next_scan_number + int((self.tomo_angle_range / self.tomo_angle_stepsize) * 8) + 10:>{padding}}\n",
+            f"{'Last scan number approx.:':<{padding}}{self.client.queue.next_scan_number + tomo_type1_total_projections + 10:>{padding}}\n",
             f"{'Current photon energy:':<{padding}}To be implemented\n",
             # f"{'Current photon energy:':<{padding}}{dev.mokev.read()['mokev']['value']:>{padding}.4f}\n",
             f"{'Exposure time:':<{padding}}{self.tomo_countingtime:>{padding}.2f}\n",

csaxs_bec/bec_ipython_client/plugins/flomni/gui_tools.py

@@ -131,9 +131,18 @@ class flomniGuiTools:
             else:
                 print("Cannot open camera_overview. Device does not exist.")
 
-            # Confirm/abort console, docked below the cameras.
+            # Confirm/abort console, docked below the cameras. The hard-stop
+            # button directly calls dev.ftransy.controller
+            # .hard_abort_and_restore_positioning_mode() (the same Galil
+            # hard-stop ftransfer_abort() uses) -- see ConsoleButtonsWidget's
+            # docstring for why this replaced a blind stop-all-devices
+            # broadcast.
             self.console = self.gui.flomni.new(
-                "ConsoleButtonsWidget", object_name="console", where="bottom"
+                "ConsoleButtonsWidget",
+                object_name="console",
+                where="bottom",
+                hard_stop_device_name="ftransy",
+                hard_stop_label="Flomni Motion Stop",
             )
             # set_layout_ratios uses relative weights, not pixels -- there is
             # no width/height kwarg on dock_area.new(). [5, 1] gives the

csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py

@@ -134,10 +134,11 @@ class OMNYTools:
             bool: True for "yes", False for "no".
 
         Note on abort: the widget's ABORT button does not write a response
-        to poll for -- it sends a real SIGINT directly to this process (see
+        to poll for -- it directly calls a configured hard motion stop and
-        ConsoleButtonsWidget._on_abort), so pressing it raises
+        sends a SIGINT to this process (see ConsoleButtonsWidget._on_abort),
-        KeyboardInterrupt here exactly as a console Ctrl+C would, and
+        so pressing it raises KeyboardInterrupt here exactly as a console
-        propagates normally out of this method without any special-casing.
+        Ctrl+C would, and propagates normally out of this method without any
+        special-casing.
         """
         if autoconfirm and default == "y":
             self.printgreen(message + " Automatically confirming default: yes")

csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons.py

@@ -4,11 +4,8 @@ import os
 import signal
 
 from bec_lib import bec_logger
-from bec_lib.endpoints import MessageEndpoints
-from bec_lib.messages import VariableMessage
 from bec_widgets import BECWidget, SafeProperty, SafeSlot
 from bec_widgets.utils.rpc_decorator import rpc_timeout
-from qtpy.QtCore import QTimer
 from qtpy.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget
 
 logger = bec_logger.logger
@@ -17,36 +14,50 @@ logger = bec_logger.logger
 class ConsoleButtonsWidget(BECWidget, QWidget):
     """
     Small Yes / No / Abort control widget, intended as a GUI replacement for
-    console prompts (e.g. ``OMNYTools.yesno()``) and as a general-purpose
+    console prompts (e.g. ``OMNYTools.yesno()``) and as a hard motion-stop
-    emergency-stop control.
+    control for a Galil-controlled beamline (flomni/OMNY/LamNI).
 
     - Yes / No: set ``response`` to "yes" / "no". A blocking CLI script can
       poll ``gui.<name>.response`` and reset it via ``clear_response()``.
-    - Abort: sends a real SIGINT to the BEC IPython client process (the
+    - Abort: calls ``queue.request_scan_abortion()`` (a safe no-op if
-      parent of the GUI server process), equivalent to pressing Ctrl+C in
+      nothing is queued, but registers a stop_id so the device server
-      the console. This works even if the client is blocked inside a motor
+      suppresses the resulting error if something else happens to be mid a
-      move or other long call, since it is a real OS signal rather than a
+      queue-tracked instruction at the same time -- this is what previously
-      polled flag. The SIGINT is only sent if the parent process actually
+      let an unrelated, un-suppressed device error kill the scan worker
-      looks like a BEC client -- it does not when the widget is opened
+      thread and require a full BEC restart), then sends a real SIGINT to
-      standalone from the launcher menu. In addition, 500 ms later a device
+      the BEC IPython client process (the parent of the GUI server
-      stop request is published to the device server (same mechanism as the
+      process), then directly calls ``hard_stop_device_name``'s
-      PositionerBox stop button): by default for ALL devices, so the widget
+      ``.controller.hard_abort_and_restore_positioning_mode()`` -- the same
-      acts as a generic emergency stop from any context, even if the client
+      Galil hard-stop (XQ#STOP,1) the CLI's ``ftransfer_abort()`` uses.
-      process is hung or dead.
+
+      SIGINT is sent BEFORE the hard stop, not after: if a client-side loop
+      is actively polling this same transfer (e.g. ftransfer_get_sample's
+      ``while True: ... if not in_progress: break``), and the hard stop's
+      own controller-level clear of ``mntprgs`` lands before that process
+      notices the interrupt, the loop would see "not in progress" and take
+      that as normal completion -- silently continuing into
+      ``ensure_gripper_up()``, which must not run mid-transfer. Sending
+      SIGINT first (an os.kill() call, effectively instant) gives the
+      target process's own ``except KeyboardInterrupt: ftransfer_abort();
+      raise FlomniError(...)`` handler a head start over the hard stop's
+      own multi-step sequence (which itself takes hundreds of ms due to
+      its internal polling/settling waits) -- without adding any
+      artificial delay of its own, so the actual motion stop is still
+      effectively immediate.
+
+    ``hard_stop_device_name`` names the device whose ``.controller`` the
+    hard stop is called on (e.g. ``"ftransy"`` for flomni). Without a
+    device configured, or if that device isn't present/enabled in the
+    current session, the button is disabled rather than silently doing
+    nothing.
     """
 
     USER_ACCESS = ["message", "message.setter", "response", "clear_response"]
     PLUGIN = True
 
     def __init__(self, parent=None, **kwargs):
-        # Devices for which a backup stop request is published when ABORT is
+        self._hard_stop_device_name = kwargs.pop("hard_stop_device_name", None)
-        # pressed. An empty list (default) means "stop ALL devices" -- the
+        self._hard_stop_label = kwargs.pop("hard_stop_label", "Motion Stop")
-        # same device-server path BEC uses on scan abort -- which makes the
-        # widget a generic emergency stop, e.g. when opened standalone from
-        # the launcher menu. Stop-all also covers the flomni sample transfer:
-        # XQ#STOP via ftransy is controller-wide and halts the #GRGET/#GRPUT
-        # thread. Pass an explicit list to restrict the stop.
-        self._backup_stop_devices = list(kwargs.pop("backup_stop_devices", []))
         super().__init__(parent=parent, **kwargs)
         self._response = ""
         # Captured once at construction time: the GUI server process is a
@@ -57,6 +68,25 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
         self._client_pid = os.getppid()
         self._init_ui()
 
+    def _hard_stop_available(self) -> bool:
+        """True if hard_stop_device_name names a device that's actually
+        present and enabled in this session -- checked once at construction
+        time (the config doesn't change mid-session).
+
+        Device access is wrapped in try/except: DeviceManagerBase.__getattr__
+        raises DeviceConfigError (not AttributeError) for an unknown device
+        name, so a plain getattr(self.dev, name, None) would NOT fall back
+        to the default and would propagate instead (same guard as
+        SampleStorageWidget._check_flomni_available()).
+        """
+        if not self._hard_stop_device_name:
+            return False
+        try:
+            device = getattr(self.dev, self._hard_stop_device_name, None)
+        except Exception:
+            return False
+        return device is not None and getattr(device, "enabled", True)
+
     def _init_ui(self):
         layout = QVBoxLayout(self)
 
@@ -81,11 +111,19 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
         self.abort_button.clicked.connect(self._on_abort)
 
         # Start with Yes/No greyed out: with no message there is nothing to
-        # respond to, so there should be nothing clickable. ABORT is left
+        # respond to, so there should be nothing clickable.
-        # always enabled -- it's an emergency stop and must work at any time,
-        # message or not.
         self._set_yesno_enabled(False)
 
+        # A dead-but-clickable "ABORT" button (no configured hard-stop
+        # target) is worse than an obviously disabled one -- disable it
+        # rather than silently doing nothing useful when pressed.
+        if self._hard_stop_available():
+            self.abort_button.setText(self._hard_stop_label)
+            self.abort_button.setEnabled(True)
+        else:
+            self.abort_button.setText("Motion Stop (not active)")
+            self.abort_button.setEnabled(False)
+
     def _set_yesno_enabled(self, enabled: bool):
         self.yes_button.setEnabled(enabled)
         self.no_button.setEnabled(enabled)
@@ -115,29 +153,42 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
 
     @SafeSlot()
     def _on_abort(self):
+        # 1) Coordinated abort first: a safe no-op if nothing is queued: when
+        #    something IS queue-tracked, this registers a real stop_id so the
+        #    device server suppresses the resulting error instead of killing
+        #    the scan worker thread (see class docstring).
+        try:
+            self.queue.request_scan_abortion()
+        except Exception:
+            logger.exception("ConsoleButtonsWidget: request_scan_abortion() failed")
+
+        # 2) SIGINT before the hard stop -- gives a client-side polling loop
+        #    (if this is a self-abort) a head start to notice the interrupt
+        #    before the hard stop's own controller-level state change could
+        #    let it exit "cleanly" instead. See class docstring for why the
+        #    order matters here. Only sent if the parent process actually
+        #    looks like a BEC client -- it does not when the widget is
+        #    opened standalone from the launcher menu.
         if self._client_is_bec_process():
             logger.warning(f"ConsoleButtonsWidget: sending SIGINT to client pid {self._client_pid}")
             os.kill(self._client_pid, signal.SIGINT)
-        else:
-            logger.warning(
-                "ConsoleButtonsWidget: parent process does not look like a BEC client;"
-                " skipping SIGINT and only sending the device stop request."
-            )
-        # Backup: direct device stop via the device server, independent of
-        # the client process. Delayed so the SIGINT-triggered abort handler
-        # in the client (which still sees mntprgs=1 and aborts in a
-        # controlled way) wins the race: if the stop landed first, #STOP
-        # would clear mntprgs and the client transfer loop would exit
-        # "cleanly" into ensure_gripper_up, which must not happen
-        # mid-transfer.
-        QTimer.singleShot(500, self._send_backup_stop)
 
-    @SafeSlot()
+        # 3) Hard motion stop -- not delayed by anything above: sending
-    def _send_backup_stop(self):
+        #    SIGINT is a near-instant os.kill() call, so this still runs
-        """Publish a stop request for the configured devices to the device server."""
+        #    effectively immediately.
-        devices = self._backup_stop_devices
+        if self._hard_stop_device_name:
-        logger.warning(f"ConsoleButtonsWidget: sending backup stop request for {devices}")
+            try:
-        self.client.connector.send(MessageEndpoints.stop_devices(), VariableMessage(value=devices))
+                device = getattr(self.dev, self._hard_stop_device_name, None)
+            except Exception:
+                device = None
+            if device is not None:
+                logger.warning(
+                    f"ConsoleButtonsWidget: hard-stopping {self._hard_stop_device_name}"
+                )
+                try:
+                    device.controller.hard_abort_and_restore_positioning_mode()
+                except Exception:
+                    logger.exception("ConsoleButtonsWidget: hard motion stop failed")
 
     @SafeProperty(str)
     def message(self):

csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py

@@ -62,6 +62,8 @@ from qtpy.QtWidgets import (
     QWidget,
 )
 
+from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
+
 logger = bec_logger.logger
 
 # ── constants ────────────────────────────────────────────────────────────────
@@ -107,6 +109,7 @@ class _SlotCell(QFrame):
         self._owner = owner
         self._occupied = False
         self._name = EMPTY_NAME
+        self._sample_owner = ""
 
         self.setFrameShape(QFrame.Shape.Box)
         self.setLineWidth(1)
@@ -126,8 +129,14 @@ class _SlotCell(QFrame):
         self._lbl_name.setAlignment(Qt.AlignmentFlag.AlignCenter)
         self._lbl_name.setWordWrap(True)
 
+        self._lbl_owner = QLabel("")
+        self._lbl_owner.setAlignment(Qt.AlignmentFlag.AlignCenter)
+        self._lbl_owner.setStyleSheet("color: #888888; font-size: 10px;")
+        self._lbl_owner.setWordWrap(True)
+
         layout.addWidget(self._lbl_num)
         layout.addWidget(self._lbl_name, stretch=1)
+        layout.addWidget(self._lbl_owner)
 
         self.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
         self.customContextMenuRequested.connect(self._show_menu)
@@ -141,16 +150,19 @@ class _SlotCell(QFrame):
             return "Gripper (100)"
         return str(self._slot)
 
-    def set_state(self, occupied: bool, name: str) -> None:
+    def set_state(self, occupied: bool, name: str, owner: str = "") -> None:
         """Update the cell's displayed state (called by the poll/refresh)."""
         self._occupied = occupied
         self._name = name if name else EMPTY_NAME
+        self._sample_owner = owner
         if occupied:
             self._lbl_name.setText(self._name)
             self._lbl_name.setStyleSheet(f"color: {COLOR_OCCUPIED};")
+            self._lbl_owner.setText(f"owner: {owner}" if owner else "")
         else:
             self._lbl_name.setText(EMPTY_LABEL)
             self._lbl_name.setStyleSheet(f"color: {COLOR_EMPTY}; font-style: italic;")
+            self._lbl_owner.setText("")
 
     # ── context menu ──────────────────────────────────────────────────────────
 
@@ -161,7 +173,7 @@ class _SlotCell(QFrame):
             act_clear = menu.addAction("Clear (set empty)")
             chosen = menu.exec_(self.mapToGlobal(pos))
             if chosen == act_rename:
-                self._owner.action_change_name(self._slot, self._name)
+                self._owner.action_change_name(self._slot, self._name, self._sample_owner)
             elif chosen == act_clear:
                 self._owner.action_clear(self._slot)
         else:
@@ -198,8 +210,8 @@ class SampleStorageWidget(BECWidget, QWidget):
         super().__init__(parent=parent, **kwargs)
         self.get_bec_shortcuts()
         self._cells: dict[int, _SlotCell] = {}
-        # last-seen (occupied, name) per slot, for change-detection in refresh()
+        # last-seen (occupied, name, owner) per slot, for change-detection in refresh()
-        self._last_state: dict[int, tuple[bool, str]] = {}
+        self._last_state: dict[int, tuple[bool, str, str]] = {}
         self._flomni_available = self._check_flomni_available()
         self._build_ui()
         self._poll_timer = QTimer(self)
@@ -229,9 +241,9 @@ class SampleStorageWidget(BECWidget, QWidget):
         """The flomni_samples device (single source of truth)."""
         return self.dev.flomni_samples
 
-    def _read_all_slots(self) -> dict[int, tuple[bool, str]]:
+    def _read_all_slots(self) -> dict[int, tuple[bool, str, str]]:
-        """Return {slot: (occupied, name)} for every slot 0–20 and the gripper
+        """Return {slot: (occupied, name, owner)} for every slot 0–20 and the
-        from a SINGLE bulk ``dev.flomni_samples.read()`` round-trip.
+        gripper from a SINGLE bulk ``dev.flomni_samples.read()`` round-trip.
 
         The per-slot accessors (``is_sample_slot_used`` +
         ``sample_names.sample{N}.get()``) would be ~44 blocking device
@@ -242,9 +254,10 @@ class SampleStorageWidget(BECWidget, QWidget):
         ``flomni_samples_sample_names_sample{N}`` /
         ``flomni_samples_sample_in_gripper`` /
         ``flomni_samples_sample_in_gripper_name``), the same keys the CLI
-        iterated over in ftransfer_sample_change.
+        iterated over in ftransfer_sample_change. The raw name signal packs
+        name+owner (see ``sample_desc_codec``), unpacked here.
         """
-        result: dict[int, tuple[bool, str]] = {}
+        result: dict[int, tuple[bool, str, str]] = {}
         try:
             data = self._samples.read()
         except Exception as exc:
@@ -260,27 +273,30 @@ class SampleStorageWidget(BECWidget, QWidget):
 
         for slot in (STAGE_SLOT, *STORAGE_SLOTS):
             used = _val(f"flomni_samples_sample_placed_sample{slot}", 0)
-            name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
+            raw_name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
-            result[slot] = (bool(used), name if name else EMPTY_NAME)
+            name, owner = unpack_desc(raw_name if raw_name else EMPTY_NAME)
+            result[slot] = (bool(used), name if name else EMPTY_NAME, owner)
 
         g_used = _val("flomni_samples_sample_in_gripper", 0)
-        g_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
+        g_raw_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
-        result[GRIPPER_SLOT] = (bool(g_used), g_name if g_name else EMPTY_NAME)
+        g_name, g_owner = unpack_desc(g_raw_name if g_raw_name else EMPTY_NAME)
+        result[GRIPPER_SLOT] = (bool(g_used), g_name if g_name else EMPTY_NAME, g_owner)
 
         return result
 
-    def _write_slot(self, slot: int, used: int, name: str) -> bool:
+    def _write_slot(self, slot: int, used: int, name: str, owner: str = "") -> bool:
         """
-        Write (used, name) to a storage slot 0–20 or the gripper, mirroring
+        Write (used, name, owner) to a storage slot 0–20 or the gripper,
-        ``Flomni.flomni_modify_storage_non_interactive()`` exactly.
+        mirroring ``Flomni.flomni_modify_storage_non_interactive()`` exactly.
         """
         try:
+            packed_name = pack_desc(name, owner)
             if slot == GRIPPER_SLOT:
                 self._samples.sample_in_gripper.set(used)
-                self._samples.sample_in_gripper_name.set(name)
+                self._samples.sample_in_gripper_name.set(packed_name)
             else:
                 getattr(self._samples.sample_placed, f"sample{slot}").set(used)
-                getattr(self._samples.sample_names, f"sample{slot}").set(name)
+                getattr(self._samples.sample_names, f"sample{slot}").set(packed_name)
             # drop the cached state for this slot so the refresh() right after
             # a write always repaints it, without waiting for the value to
             # differ from a possibly-stale cache entry
@@ -374,17 +390,17 @@ class SampleStorageWidget(BECWidget, QWidget):
         state = self._read_all_slots()
         if not state:
             return  # bulk read failed; leave the current display untouched
-        for slot, (occupied, name) in state.items():
+        for slot, (occupied, name, owner) in state.items():
-            if self._last_state.get(slot) == (occupied, name):
+            if self._last_state.get(slot) == (occupied, name, owner):
                 continue
             cell = self._cells.get(slot)
             if cell is not None:
-                cell.set_state(occupied, name)
+                cell.set_state(occupied, name, owner)
-            self._last_state[slot] = (occupied, name)
+            self._last_state[slot] = (occupied, name, owner)
 
     # ── mutation actions (called from _SlotCell context menu) ─────────────────
 
-    def action_change_name(self, slot: int, current_name: str) -> None:
+    def action_change_name(self, slot: int, current_name: str, current_owner: str = "") -> None:
         """Rename an occupied slot (stays occupied)."""
         name, ok = QInputDialog.getText(
             self,
@@ -403,7 +419,14 @@ class SampleStorageWidget(BECWidget, QWidget):
                 "Use “Clear (set empty)” to empty the slot instead.",
             )
             return
-        if self._write_slot(slot, 1, name):
+        owner, ok = QInputDialog.getText(
+            self,
+            "Sample owner",
+            f"Owner for slot {self._slot_title(slot)} (optional):",
+            text=current_owner,
+        )
+        owner = owner.strip() if ok else current_owner
+        if self._write_slot(slot, 1, name, owner):
             self.refresh()
 
     def action_new_sample(self, slot: int) -> None:
@@ -417,7 +440,11 @@ class SampleStorageWidget(BECWidget, QWidget):
         if not name or name == EMPTY_NAME:
             QMessageBox.warning(self, "Invalid name", "Please enter a non-empty sample name.")
             return
-        if self._write_slot(slot, 1, name):
+        owner, ok = QInputDialog.getText(
+            self, "Sample owner", f"Owner for slot {self._slot_title(slot)} (optional):"
+        )
+        owner = owner.strip() if ok else ""
+        if self._write_slot(slot, 1, name, owner):
             self.refresh()
 
     def action_clear(self, slot: int) -> None:

csaxs_bec/devices/omny/flomni_sample_storage.py

@@ -6,6 +6,8 @@ from ophyd import DynamicDeviceComponent as Dcpt
 from ophyd import EpicsSignal
 from prettytable import PrettyTable
 
+from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
+
 
 class FlomniSampleStorageError(Exception):
     pass
@@ -19,6 +21,9 @@ class FlomniSampleStorage(Device):
         "unset_sample_slot",
         "set_sample_in_gripper",
         "unset_sample_in_gripper",
+        "get_sample_name",
+        "get_sample_owner",
+        "get_sample_name_and_owner",
         "show_all",
     ]
     SUB_VALUE = "value"
@@ -54,12 +59,12 @@ class FlomniSampleStorage(Device):
         self.wait_for_connection()
         self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
 
-    def set_sample_slot(self, slot_nr: int, name: str) -> bool:
+    def set_sample_slot(self, slot_nr: int, name: str, owner: str = "") -> bool:
         if slot_nr > 20:
             raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
 
         getattr(self.sample_placed, f"sample{slot_nr}").set(1)
-        getattr(self.sample_names, f"sample{slot_nr}").set(name)
+        getattr(self.sample_names, f"sample{slot_nr}").set(pack_desc(name, owner))
 
     def unset_sample_slot(self, slot_nr: int) -> bool:
         if slot_nr > 20:
@@ -68,9 +73,9 @@ class FlomniSampleStorage(Device):
         getattr(self.sample_placed, f"sample{slot_nr}").set(0)
         getattr(self.sample_names, f"sample{slot_nr}").set("-")
 
-    def set_sample_in_gripper(self, name: str) -> bool:
+    def set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
         self.sample_in_gripper.set(1)
-        self.sample_in_gripper_name.set(name)
+        self.sample_in_gripper_name.set(pack_desc(name, owner))
 
     def unset_sample_in_gripper(self) -> bool:
         self.sample_in_gripper.set(0)
@@ -86,7 +91,15 @@ class FlomniSampleStorage(Device):
 
     def get_sample_name(self, slot_nr) -> str:
         val = getattr(self.sample_names, f"sample{slot_nr}").get()
-        return str(val)
+        return unpack_desc(str(val))[0]
+
+    def get_sample_owner(self, slot_nr) -> str:
+        val = getattr(self.sample_names, f"sample{slot_nr}").get()
+        return unpack_desc(str(val))[1]
+
+    def get_sample_name_and_owner(self, slot_nr) -> tuple:
+        val = getattr(self.sample_names, f"sample{slot_nr}").get()
+        return unpack_desc(str(val))
 
     def show_all(self):
         t = PrettyTable()
@@ -105,13 +118,19 @@ class FlomniSampleStorage(Device):
         print("\n\nFollowing samples are currently loaded:\n")
         for ct in range(1, 21):
             if self.is_sample_slot_used(ct):
-                print(f"  Position {ct:2.0f}: {self.get_sample_name(ct)}")
+                name, owner = self.get_sample_name_and_owner(ct)
+                owner_suffix = f"  (owner: {owner})" if owner else ""
+                print(f"  Position {ct:2.0f}: {name}{owner_suffix}")
         if self.sample_in_gripper.get():
-            print(f"\n  Gripper:        {self.sample_in_gripper_name.get()}\n")
+            name, owner = unpack_desc(str(self.sample_in_gripper_name.get()))
+            owner_suffix = f"  (owner: {owner})" if owner else ""
+            print(f"\n  Gripper:        {name}{owner_suffix}\n")
         else:
             print(f"\n  Gripper:        no sample\n")
 
         if self.is_sample_slot_used(0):
-            print(f"  flOMNI stage:   {self.get_sample_name(0)}\n")
+            name, owner = self.get_sample_name_and_owner(0)
+            owner_suffix = f"  (owner: {owner})" if owner else ""
+            print(f"  flOMNI stage:   {name}{owner_suffix}\n")
         else:
             print(f"  flOMNI stage:   no sample\n")

csaxs_bec/devices/omny/galil/fgalil_ophyd.py

@@ -41,6 +41,7 @@ class FlomniGalilController(GalilController):
         "lights_off",
         "lights_on",
         "print_command_history",
+        "hard_abort_and_restore_positioning_mode",
     ]
 
     def is_axis_moving(self, axis_Id, axis_Id_numeric) -> bool:

csaxs_bec/devices/omny/galil/galil_ophyd.py

@@ -46,6 +46,7 @@ class GalilController(Controller):
         "is_thread_active",
         "all_axes_referenced",
         "print_command_history",
+        "hard_abort_and_restore_positioning_mode",
     ]
 
     OKBLUE = "\033[94m"
@@ -100,6 +101,37 @@ class GalilController(Controller):
         else:
             return ":"
 
+    def hard_abort_and_restore_positioning_mode(
+        self, transfer_thread_id: int = 3, timeout: float = 5.0
+    ) -> None:
+        """
+        Hard abort of a running sample-transfer routine: stops all axes and
+        threads on the controller (XQ#STOP,1 via stop_all_axes(), which
+        halts the transfer thread, aborts all motion, and clears
+        mntprgs/mntmod), waits for the transfer thread to actually halt,
+        then switches the controller back to positioning mode (#POSMODE
+        refuses while mntprgs=1, so this must wait first).
+
+        Shared by both the CLI (Flomni.ftransfer_abort(), via Ctrl+C) and
+        the GUI hard-stop button, so there is exactly one implementation of
+        this safety-critical stop sequence -- two independent
+        implementations previously drifting apart is what let the GUI path
+        end up unsafe.
+        """
+        self.stop_all_axes()
+        start = time.time()
+        while self.is_thread_active(transfer_thread_id):
+            if time.time() - start > timeout:
+                raise GalilError(
+                    f"Hard abort requested but transfer thread {transfer_thread_id} did not "
+                    f"stop within {timeout} s. Check the controller."
+                )
+            time.sleep(0.1)
+        self.socket_put_confirmed("XQ#POSMODE")
+        time.sleep(0.5)
+        if bool(float(self.socket_put_and_receive("MG mntmod").strip())):
+            raise GalilError("System is still in mount mode after hard abort.")
+
     def get_digital_input(self, channel):
         return bool(float(self.socket_put_and_receive(f"MG @IN[{channel}]").strip()))
 

csaxs_bec/devices/omny/sample_desc_codec.py

@@ -0,0 +1,52 @@
+"""Pack/unpack a sample name + owner into a single EPICS DESC field.
+
+Sample storage (flomni_sample_storage.py, omny_sample_storage.py) has only
+one free-text field per slot -- the underlying record's DESC field, used as
+the sample name. There is no separate PV for an owner, so an owner has to be
+packed into that same string.
+
+DESC_MAX_LEN is assumed (~40 chars is the generic EPICS base-record DESC
+size), not confirmed against the real IOC .db (not present in this repo) --
+verify against the real IOC before production rollout; the sim's mocked PVs
+do not enforce any length limit.
+"""
+
+DESC_MAX_LEN = 40
+EMPTY_SENTINEL = "-"
+DELIMITER = " | "
+
+
+def pack_desc(name: str, owner: str = "") -> str:
+    """Pack (name, owner) into a single DESC string.
+
+    The empty-slot sentinel ("-") always passes through untouched, never
+    gets an owner appended. If the combined string doesn't fit
+    DESC_MAX_LEN, the name is truncated first (not the owner): owner is
+    typically a short, fixed-format identifier (e.g. an e-account) that
+    matters for accountability, while name is free text that already
+    tolerates truncation better.
+    """
+    if name == EMPTY_SENTINEL:
+        return EMPTY_SENTINEL
+    if not owner:
+        return name[:DESC_MAX_LEN]
+    combined = f"{name}{DELIMITER}{owner}"
+    if len(combined) <= DESC_MAX_LEN:
+        return combined
+    name_budget = max(0, DESC_MAX_LEN - len(DELIMITER) - len(owner))
+    return f"{name[:name_budget]}{DELIMITER}{owner}"[:DESC_MAX_LEN]
+
+
+def unpack_desc(raw: str) -> tuple[str, str]:
+    """Unpack a raw DESC string into (name, owner).
+
+    Returns owner="" for a legacy string with no delimiter (no owner was
+    ever recorded), and passes the "-" empty-slot sentinel through
+    untouched with owner="".
+    """
+    if raw is None or raw == EMPTY_SENTINEL:
+        return (EMPTY_SENTINEL, "")
+    if DELIMITER not in raw:
+        return (raw, "")
+    name, _, owner = raw.partition(DELIMITER)
+    return (name, owner)

omny_e2e_tests/sim_flomni_harness.py

@@ -417,6 +417,37 @@ def main():  # noqa: C901
         "slot 1 -> gripper bookkeeping",
     )
 
+    # --- 9b. sample owner (packed into the same DESC field as name) ------------------
+    samples.set_sample_slot(2, "owner_test_sample", owner="mholler")
+    check(
+        samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns plain name only"
+    )
+    check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the packed owner")
+    check(
+        samples.get_sample_name_and_owner(2) == ("owner_test_sample", "mholler"),
+        "get_sample_name_and_owner returns (name, owner)",
+    )
+    samples.unset_sample_slot(2)
+    check(
+        str(getattr(samples.sample_names, "sample2").get()) == "-",
+        "unset_sample_slot writes bare '-', not '- | <owner>'",
+    )
+    # gripper move (slot -> gripper) must preserve owner, not just name. The
+    # gripper is a separate Cpt (sample_in_gripper_name), not part of the
+    # sample_names slot dict, so unpack it directly rather than via
+    # get_sample_name_and_owner (which only covers slots 0-20).
+    from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
+
+    samples.set_sample_slot(4, "transfer_test_sample", owner="e12345")
+    name, owner = samples.get_sample_name_and_owner(4)
+    samples.set_sample_in_gripper(name, owner=owner)
+    samples.unset_sample_slot(4)
+    check(
+        unpack_desc(str(samples.sample_in_gripper_name.get())) == ("transfer_test_sample", "e12345"),
+        "owner survives a slot -> gripper move",
+    )
+    samples.unset_sample_in_gripper()
+
     frame_a = ids_cam.cam.get_image_data()
     frame_b = ids_cam.cam.get_image_data()
     check(not np.array_equal(frame_a, frame_b), "IDS frames are live (noise explicitly enabled)")

tests/tests_bec_ipython_client/test_flomni_tomo_angles.py

@@ -0,0 +1,121 @@
+import numpy as np
+import pytest
+
+from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni
+
+plan = Flomni._subtomo_angle_plan
+
+STEPSIZES = [10.0, 7.0, 25.0, 12.5]
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+def test_180_mode_regression(stepsize):
+    """180-degree mode must be byte-for-byte identical to the original
+    scheme: all 8 sub-tomograms interlace the same [0,180) span at spacing
+    step/8, with no duplicates and no gaps."""
+    all_angles = []
+    for n in range(1, 9):
+        angles, offset, N, step = plan(n, 180, stepsize)
+        assert offset == 0
+        assert len(angles) == N
+        all_angles.append(angles)
+
+    combined = np.sort(np.concatenate(all_angles))
+    assert len(combined) == len(np.unique(np.round(combined, 6)))
+    diffs = np.diff(combined)
+    _, _, expected_total = _actual_grid(stepsize)
+    expected_step = 180.0 / (expected_total)
+    assert np.allclose(diffs, expected_step, atol=1e-6)
+    assert combined.min() >= 0
+    assert combined.max() < 180
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+def test_360_mode_no_duplicates_and_no_internal_180_pairs(stepsize):
+    """360-degree mode: no sub-tomogram may, by itself, contain two angles
+    exactly 180 degrees apart (that redundancy was the bug), and the
+    combined 8-sub-tomogram set must contain no duplicate angles."""
+    all_angles = []
+    for n in range(1, 9):
+        angles, offset, N, step = plan(n, 360, stepsize)
+        assert offset == 0
+        assert len(angles) == N
+        assert angles.max() - angles.min() < 180
+        setc = set(np.round(angles, 6))
+        for a in setc:
+            assert round((a + 180) % 360, 6) not in setc
+        all_angles.append(angles)
+
+    combined = np.sort(np.concatenate(all_angles))
+    assert len(combined) == len(np.unique(np.round(combined, 6)))
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+def test_total_projection_count_identical_180_vs_360(stepsize):
+    """Total projections for a given tomo_angle_stepsize must be the same
+    whether tomo_angle_range is 180 or 360 (confirmed requirement)."""
+    _, _, n180 = _actual_grid(stepsize)
+    total_360 = sum(len(plan(n, 360, stepsize)[0]) for n in range(1, 9))
+    assert total_360 == n180
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+def test_360_mode_full_grid_evenly_spaced(stepsize):
+    """All 8 sub-tomograms combined must be a complete, evenly-spaced grid
+    covering the full [0,360) range at spacing step/4, no gaps."""
+    _, step180, _ = _actual_grid(stepsize)
+    all_angles = np.sort(np.concatenate([plan(n, 360, stepsize)[0] for n in range(1, 9)]))
+    diffs = np.diff(all_angles)
+    assert np.allclose(diffs, step180 / 4.0, atol=1e-6)
+    assert all_angles.min() >= 0
+    assert all_angles.max() < 360
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+@pytest.mark.parametrize("pair", [(1, 2), (3, 4), (5, 6), (7, 8)])
+def test_360_mode_adjacent_pair_is_complete_coarse_tomogram(stepsize, pair):
+    """Any adjacent pair (1,2)/(3,4)/(5,6)/(7,8) must independently
+    reconstruct a complete, evenly-spaced 360-degree tomogram at the
+    coarse (unrefined) step spacing."""
+    _, step180, _ = _actual_grid(stepsize)
+    a, b = pair
+    combined = np.sort(np.concatenate([plan(a, 360, stepsize)[0], plan(b, 360, stepsize)[0]]))
+    diffs = np.diff(combined)
+    assert np.allclose(diffs, step180, atol=1e-6)
+    assert combined.min() >= 0
+    assert combined.max() < 360
+
+
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+@pytest.mark.parametrize("quartet", [(1, 2, 3, 4), (5, 6, 7, 8)])
+def test_360_mode_quartet_is_complete_half_sampled_tomogram(stepsize, quartet):
+    """Either quartet (1-4 or 5-8) must independently reconstruct a
+    complete, evenly-spaced 360-degree tomogram at half the finest
+    (8-subtomo) spacing."""
+    _, step180, _ = _actual_grid(stepsize)
+    combined = np.sort(np.concatenate([plan(n, 360, stepsize)[0] for n in quartet]))
+    diffs = np.diff(combined)
+    assert np.allclose(diffs, step180 / 2.0, atol=1e-6)
+    assert combined.min() >= 0
+    assert combined.max() < 360
+
+
+@pytest.mark.parametrize("tomo_angle_range", [180, 360])
+@pytest.mark.parametrize("stepsize", STEPSIZES)
+def test_resume_round_trip(tomo_angle_range, stepsize):
+    """Resuming with an explicit start_angle must recover the exact loop
+    index that originally produced that angle, for every sub-tomogram and
+    every position -- including the float tie-break cases (subtomo 2 in
+    180 mode; subtomos 3 and 4 in 360 mode, both landing on phase/step ==
+    0.5)."""
+    for n in range(1, 9):
+        angles, _, N, _ = plan(n, tomo_angle_range, stepsize)
+        for i in range(N):
+            _, offset, _, _ = plan(n, tomo_angle_range, stepsize, start_angle=angles[i])
+            assert offset == i, f"n={n} i={i} range={tomo_angle_range} stepsize={stepsize}"
+
+
+def _actual_grid(stepsize):
+    N = int(180.0 / stepsize)
+    step = 180.0 / N
+    return N, step, N * 8

tests/tests_bec_ipython_client/test_sample_desc_codec.py

@@ -0,0 +1,59 @@
+import pytest
+
+from csaxs_bec.devices.omny.sample_desc_codec import DESC_MAX_LEN, pack_desc, unpack_desc
+
+
+def test_round_trip():
+    packed = pack_desc("my_sample", "mholler")
+    assert packed == "my_sample | mholler"
+    assert unpack_desc(packed) == ("my_sample", "mholler")
+
+
+def test_no_owner_no_trailing_delimiter():
+    """pack_desc(name, "") must equal name exactly -- required so the
+    raw-copy call sites (e.g. gripper<->slot transfer) stay no-ops when no
+    owner has ever been set."""
+    assert pack_desc("my_sample") == "my_sample"
+    assert pack_desc("my_sample", "") == "my_sample"
+
+
+def test_legacy_string_no_delimiter():
+    """A pre-existing sample name with no delimiter unpacks to (name, "")."""
+    assert unpack_desc("my_sample") == ("my_sample", "")
+
+
+def test_empty_sentinel_round_trip():
+    assert pack_desc("-") == "-"
+    assert pack_desc("-", "someone") == "-"  # never pack an owner onto the sentinel
+    assert unpack_desc("-") == ("-", "")
+    assert unpack_desc(None) == ("-", "")
+
+
+@pytest.mark.parametrize("owner", ["", "mholler"])
+def test_truncation_long_name(owner):
+    long_name = "x" * 60
+    packed = pack_desc(long_name, owner)
+    assert len(packed) <= DESC_MAX_LEN
+    name, unpacked_owner = unpack_desc(packed)
+    assert unpacked_owner == owner
+    assert name == long_name[: len(name)]  # truncated prefix of the original name
+
+
+def test_truncation_long_owner():
+    packed = pack_desc("s", "y" * 60)
+    assert len(packed) <= DESC_MAX_LEN
+
+
+def test_truncation_no_owner_no_wasted_delimiter():
+    """Truncating a too-long name with no owner must not waste characters
+    on a trailing delimiter that has nothing after it."""
+    packed = pack_desc("x" * 60, "")
+    assert packed == "x" * DESC_MAX_LEN
+    assert not packed.endswith(" | ")
+
+
+def test_pack_stays_within_max_len_for_typical_values():
+    for name_len in range(0, 45, 5):
+        for owner_len in range(0, 20, 5):
+            packed = pack_desc("n" * name_len, "o" * owner_len if owner_len else "")
+            assert len(packed) <= DESC_MAX_LEN