feat(omny): port tomo-queue backend from flomni/lamni (Phase 1 of 2)

Mirko wants the tomo params GUI widget available for OMNY before running
real tomo scans. That widget (Phase 2, separate follow-up) needs a working
job queue and scan progress visible outside the CLI session that started
it -- neither existed for OMNY. OMNY now inherits TomoQueueMixin
(OMNY_shared/tomo_queue_mixin.py), the same shared backend flomni/lamni
already use, giving it tomo_queue_add/delete/move/clear/show/execute/
reacquire, at_each_angle_hook registration, and tomo_scan_resume() for the
first time.

self.progress is now a global-var-backed proxy (_ProgressProxy, byte-for-
byte port of flomni's/lamni's own), not a plain in-memory dict -- the
load-bearing change the queue mixin and the eventual widget's busy-banner
both depend on. tomo_queue_mixin.py's tomo_queue_reacquire() hardcoded
tomo_type == 1; generalized via a new _TOMO_TYPE_1_LIKE class attribute
(default frozenset({1}), flomni/lamni unaffected) since OMNY has three
"equally spaced sub-tomograms" flavors (types 1/4/5), not just one.

Along the way, fixed several real bugs in tomo_scan()/write_pdf_report()
that predate this session and would have crashed on a real account/
real energy read (bec.active_account.decode() -- active_account is str
not bytes; dev.mokev doesn't exist for OMNY, only dev.ccm_energy; an
unguarded logbook send; a stale "LamNI" logo/tag copy-paste leftover) --
all direct ports of fixes flomni/lamni already needed and got. Also added
the missing unconditional omnygui_show_progress() call and heartbeat-
clearing try/finally in tomo_scan(), matching both.

Verification: 23 new tests (test_omny_tomo_queue.py, test_omny_tomo_scan.py)
covering the mixin wiring, hook dispatch/resolution, tomo_scan_resume()/
_resolve_type1_projection()/reacquire() across all three sub-tomogram
flavors, account handling, and heartbeat clearing on both normal completion
and a mid-scan exception. Full suite: 702 passed. Not yet live-tested
against a running session (the local bec-server may be in active use by
Mirko's own parallel session) -- next step is his own pull-and-test.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLrD7sVYGLAzsQjLVJpCgt

csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py

@@ -27,10 +27,15 @@ Consumers (e.g. ``Flomni``, ``LamNI``) must:
       ``flomni_at_each_angle`` builtins hook, single-point acquisition -- that
       it stays out of this mixin)
     - implement their own ``_resolve_type1_projection()`` (dispatched
-      generically by ``tomo_queue_reacquire()``), mapping a flat tomo_type-1
-      projection number to a ``(subtomo_start, start_angle)`` pair -- the
-      angle math is setup-specific (see each setup's own
-      ``_subtomo_angle_plan()``), same reasoning as ``tomo_scan()`` above
+      generically by ``tomo_queue_reacquire()`` for any tomo_type in
+      ``_TOMO_TYPE_1_LIKE``), mapping a flat projection number to a
+      ``(subtomo_start, start_angle)`` pair -- the angle math is
+      setup-specific (see each setup's own ``_subtomo_angle_plan()``/
+      ``_equally_spaced_subtomo_angles()``), same reasoning as
+      ``tomo_scan()`` above
+    - optionally override ``_TOMO_TYPE_1_LIKE`` (default ``frozenset({1})``)
+      if their setup has more than one "equally spaced sub-tomograms"
+      tomo_type flavor (e.g. OMNY's types 1/4/5 -- 2/4/8 sub-tomograms)
 """
 
 from __future__ import annotations
@@ -210,6 +215,13 @@ class TomoQueueMixin:
     _TOMO_QUEUE_ACTIONS: dict = {}
     _TOMO_QUEUE_MOVE_DEVICES: dict = {}
 
+    # tomo_type values that behave like "type 1" for tomo_queue_reacquire()'s
+    # purposes -- i.e. resolved via _resolve_type1_projection() rather than a
+    # flat projection_number. Default matches flomni/lamni (type 1 only, their
+    # only "equally spaced sub-tomograms" flavor); OMNY overrides this to
+    # {1, 4, 5} since it has three such flavors (2/4/8 sub-tomograms).
+    _TOMO_TYPE_1_LIKE: frozenset = frozenset({1})
+
     # Name of a registered at_each_angle hook (see register_at_each_angle_hook()),
     # or None for the default per-projection behaviour.
     at_each_angle_hook = _GlobalVarParam(None)
@@ -1012,10 +1024,9 @@ class TomoQueueMixin:
             job_index: Position of the job to reopen, in the *current*
                 queue order (same indexing as ``tomo_queue_show()``/
                 ``tomo_queue_delete()``).
-            projection_number: Flat 0-indexed projection to resume from,
-                for any tomo_type (1, 2, or 3) -- see
-                ``_resolve_type1_projection()`` for how this maps onto a
-                tomo_type-1 job's (subtomo, angle).
+            projection_number: Flat 0-indexed projection to resume from, for
+                any tomo_type -- see ``_resolve_type1_projection()`` for how
+                this maps onto a ``_TOMO_TYPE_1_LIKE`` job's (subtomo, angle).
 
         Raises:
             TomoQueueError: ``job_index`` is out of range, the job at that
@@ -1049,7 +1060,7 @@ class TomoQueueMixin:
         for name, value in job["params"].items():
             setattr(self, name, value)
 
-        if self.tomo_type == 1:
+        if self.tomo_type in self._TOMO_TYPE_1_LIKE:
             subtomo_start, start_angle = self._resolve_type1_projection(projection_number)
             self.progress["subtomo"] = subtomo_start
             self.progress["angle"] = start_angle

csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/OPEN_ISSUES.md

@@ -26,6 +26,12 @@ removed" below. Tomo type 3 also gained a hard cap of 2.5 degrees on the angular
 only a minimum-100-projections floor, which happened to already satisfy this but wasn't expressed
 as a wear limit).
 
+**Just landed**: the tomo-queue backend (Phase 1 of 2 of porting the tomo params GUI widget to
+OMNY) — `OMNY` now has a real job queue, `at_each_angle_hook` registration, `tomo_scan_resume()`,
+and scan progress visible outside the CLI session (previously a plain in-memory dict). See "Tomo-
+queue backend ported to OMNY" below. Phase 2 (the widget itself) is a separate follow-up, not yet
+started — not live-tested against a running session yet either; that's next, once Mirko pulls.
+
 **Still open / deferred** (unchanged from before, not touched this pass):
 - `x_ray_eye_align.py` is still LamNI-derived, unadapted code beyond the two call-site renames
   that unblocked the GUI-open crash — see its own section below. The GUI itself opens fine now;
@@ -33,6 +39,7 @@ as a wear limit).
 - The full gripper sample-transfer chain (`otransfer_put_sample()`/`otransfer_get_sample()`),
   dewar/temperature methods, sample storage bookkeeping, and `omny_alignment_mixin.py`/
   `omny_optics_mixin.py` beyond what scans exercise — see "Checklist" section below.
+- The `TomoParamsWidget`/`OMNY_TomoParams` GUI widget itself (Phase 2 of the tomo-queue port).
 
 ## Standing question: OMNY should follow flOMNI, not LamNI
 
@@ -126,24 +133,29 @@ Still untested (not attempted yet, not known-broken — just unexplored):
 - `omny_webpage_generator.py` (2957 lines) is dead/unwired code — never instantiated by
   `omny.py`, per `FLOMNI_TO_LAMNI_COMPARISON.md`'s note ("omny was left untouched...
   separate future task if omny ever needs a real one"). Not touched this session either.
-- The tomo-queue / `TomoParamsWidget` generalization done for LamNI
-  (`LamNI/AI_docs/TOMO_QUEUE_PORT.md`, `TOMO_PARAMS_GUI_PORT.md`) has no OMNY counterpart
-  yet — out of scope for the sim-validation work, in scope for the eventual macro-
-  standardization pass.
+- **Tomo-queue backend: ported, Phase 1 of 2 (this session).** `OMNY` now inherits
+  `TomoQueueMixin` (`OMNY_shared/tomo_queue_mixin.py`), the same shared backend flomni/lamni
+  use — `tomo_queue_add/delete/move/clear/show/execute/reacquire`,
+  `register_at_each_angle_hook`/`at_each_angle_hook`, `tomo_scan_resume()`. See "Tomo-queue
+  backend ported to OMNY" below for the full detail. **Not yet done (Phase 2, separate
+  follow-up)**: the `TomoParamsWidget`/`OMNY_TomoParams` GUI widget itself
+  (`bec_widgets/widgets/tomo_params/tomo_params.py`) still has no `SETUP_PROFILES["omny"]`
+  entry and its tomo-type selector/job-display logic is hardcoded to 3 tomo types (flomni's
+  and lamni's), not OMNY's 5 — needs its own plan once Phase 1 is live-tested.
 
-## gui_tools.py progress fields (deferred on purpose)
+## gui_tools.py progress ETA/heartbeat line (still not rendered, data now exists)
 
 `gui_tools.py` was modernized to flomni's current single-level `.new()` / `_is_deleted()`-guard
 pattern (window lifecycle, xray-eye-align GUI, cameras + hard-stop console, `RingProgressBar`).
-One piece was deliberately **not** ported: flomni's `self.progress` dict carries
-`tomo_start_time`/`heartbeat`/`estimated_remaining_time`/`estimated_finish_time` plus a
-`_describe_active_hook()` method feeding an ETA/heartbeat line in `_flomnigui_update_progress()`.
-omny's `self.progress` (set up in `OMNY.__init__`, `omny.py`) still only has the original 7 fields
-(`subtomo`, `subtomo_projection`, `subtomo_total_projections`, `projection`,
-`total_projections`, `angle`, `tomo_type`) and `_omnygui_update_progress()` was written against
-just those. Mirko's call: keep it simple for this pass, revisit later. Not a crash risk —
-`omny_webpage_generator.py` already reads the ETA-ish fields defensively via `.get()`, degrading
-to "N/A" when absent, and that file is unwired/dead code anyway (see below).
+As of the tomo-queue backend port (see below), `self.progress` now DOES carry
+`tomo_start_time`/`heartbeat`/`estimated_remaining_time`/`estimated_finish_time` (global-var-backed,
+same shape as flomni's/lamni's) — what's still not ported is purely a `gui_tools.py` rendering
+detail: `_omnygui_update_progress()` was never updated to display an ETA/heartbeat line the way
+flomni's `_flomnigui_update_progress()`/`_describe_active_hook()` do, and still only builds its
+text from the original 7 fields. Not a crash risk either way — `_omnygui_update_progress()`
+ignores fields it doesn't read, and `omny_webpage_generator.py` (unwired/dead code, see below)
+already reads the ETA-ish fields defensively via `.get()`, degrading to "N/A" when absent. Revisit
+whenever the progress ring GUI itself gets an ETA line, independent of the Phase 2 widget work.
 
 ## cam_xeye device (simulated only, real hardware config still needed)
 
@@ -274,6 +286,92 @@ Test coverage: `tests/tests_bec_ipython_client/test_omny_tomo_angles.py`'s
 `"x12sa"` canned answer; new
 `test_tomo_parameters_clamps_type_3_step_size_to_max_2_5_degrees` covers the clamp.
 
+## Tomo-queue backend ported to OMNY (Phase 1 of 2)
+
+Mirko wants the tomo params GUI widget available for OMNY before starting real tomo scans. That
+widget (Phase 2, separate follow-up) depends on a working tomo-queue backend and on scan progress
+being visible outside the CLI session that started it -- neither existed for OMNY before this pass.
+Same shared infrastructure and pattern as the flomni -> lamni port
+(`LamNI/AI_docs/TOMO_QUEUE_PORT.md`, `TOMO_PARAMS_GUI_PORT.md`), done as the same two-phase split
+per Mirko's explicit choice.
+
+**`OMNY` now inherits `TomoQueueMixin`** (`OMNY_shared/tomo_queue_mixin.py`, first in the MRO,
+matching flomni/lamni), giving it `tomo_queue_add/add_command/delete/move/clear/show/execute/
+resume/reacquire`, `register_at_each_angle_hook`/`unregister_at_each_angle_hook`/
+`list_at_each_angle_hooks`/`at_each_angle_hook` for the first time. New class attributes:
+`_TOMO_SCAN_PARAM_NAMES` (mirrors flomni's names exactly -- OMNY's properties already mirror
+flomni's 1:1, no `tomo_angle_range`/`single_point_*`/`zero_deg_reference_at_each_subtomo`/
+`frames_per_trigger` since none of these exist as OMNY properties), `_TOMO_QUEUE_MOVE_DEVICES`
+(`ccm_energy`+`idgap`, **not** `mokev` -- OMNY has no `mokev` device in any config, real or
+simulated), `_TOMO_QUEUE_ACTIONS` (same `move`/`optimize_idgap` registry as flomni/lamni,
+`optimize_idgap` still a stub).
+
+**`self.progress` is now a global-var-backed proxy, not a plain dict.** Before this pass, OMNY's
+`self.progress = {}` was a plain in-memory instance dict -- invisible to any other BEC client
+session, a GUI process included. New `_ProgressProxy` class (direct, byte-for-byte port of
+flomni's/lamni's own, backed by `client.get_global_var("tomo_progress")`, same global-var key --
+only one tomography setup is ever live per BEC session, no collision risk), with `progress` now a
+`@property`. Carries the original 7 fields (`subtomo`, `subtomo_projection`,
+`subtomo_total_projections`, `projection`, `total_projections`, `angle`, `tomo_type`) plus
+`tomo_start_time`/`estimated_remaining_time`/`estimated_finish_time`/`heartbeat`/
+`accumulated_idle_time`, matching flomni's/lamni's `_DEFAULTS` shape.
+
+**`tomo_queue_mixin.py`'s `tomo_queue_reacquire()` generalized (small, additive, behavior-
+preserving change to shared code)**: it hardcoded `if self.tomo_type == 1:` to decide when to
+resolve via `_resolve_type1_projection()`. OMNY has three "equally spaced sub-tomograms" flavors
+(types 1/4/5 -- 2/4/8 sub-tomograms), not just type 1. Added `_TOMO_TYPE_1_LIKE: frozenset`
+(default `frozenset({1})`, so flomni/lamni are unaffected -- confirmed via their existing test
+suites, unchanged and passing), which OMNY overrides to `frozenset({1, 4, 5})`.
+
+**New OMNY methods, direct ports generalized for 3 sub-tomogram counts**: `tomo_scan_resume()`
+(dispatches subtomo/angle for tomo_type in `_TOMO_TYPE_1_LIKE`, projection_number for 2/3) and
+`_resolve_type1_projection()` (reuses the already-verified `_equally_spaced_subtomo_angles()` --
+does not re-derive the angle grid a second way).
+
+**Real, pre-existing bugs fixed along the way** (same shapes flomni/lamni already needed and got,
+found by reading `tomo_scan()`/`write_pdf_report()` against the mixin's needs, not newly
+introduced):
+- `bec.active_account.decode()` in `tomo_scan()` -- `active_account` is a plain `str`, not
+  `bytes`; crashed on every real account. Removed (OMNY's existing `if bec.active_account != "":
+  ... else: tomo_id = 0` branch structure kept as-is, unlike flomni's/lamni's "always attempt,
+  let add_sample_database() decide" pattern -- OMNY's own `add_sample_database()` has no
+  test-server/empty-account fallback to lean on).
+- `dev.mokev.read()['mokev']['value']` in `write_pdf_report()` -- OMNY has no `mokev` device in
+  any config. Fixed to `dev.ccm_energy` (from `bl_optics_hutch.yaml`, always included regardless
+  of active endstation) with a try/except -> `"N/A"` fallback, matching lamni's identical fix.
+  Confirmed `ccm_energy` is *not* in the local simulated OMNY config either (`simulated_omny.yaml`/
+  `simulated_bl_endstation.yaml` have no such device), so live sim testing will exercise the
+  fallback path, not a real energy read -- still correct/safe behavior, just not the success path.
+- `bec.logbook.LogbookMessage()`/`send_logbook_message()` in `write_pdf_report()` was unguarded --
+  wrapped in try/except, matching `write_to_scilog()`'s existing tolerance one method up.
+- `write_pdf_report()`'s scilog logo/tags still said `"LamNI"`/pointed at a nonexistent
+  `LamNI_logo.png` (omny's plugin directory only has `OMNY.png`) -- `msg.add_file(logo_path)`
+  would have raised unguarded (now inside the same try/except above regardless). Fixed to
+  `OMNY.png`/`"OMNY"`.
+- `tomo_scan()` never called `self.omnygui_show_progress()` unconditionally at scan start (only
+  per-angle inside `_tomo_scan_at_angle()`) and had no try/finally to clear `heartbeat` on exit --
+  both fixed, matching flomni's/lamni's identical pattern.
+- `_at_each_angle()` now checks `self.at_each_angle_hook` (mixin-provided) first, falling through
+  to the pre-existing `omny_at_each_angle` builtins-global hook, then the default
+  `tomo_scan_projection()` + `tomo_reconstruct()` -- same 3-tier structure as flomni's, minus its
+  single-point branch (OMNY has none).
+
+**Verification**: new `tests/tests_bec_ipython_client/test_omny_tomo_queue.py` (17 tests --
+param-name snapshotting, queue add/show/execute/resume/reacquire, add-command move validation,
+`at_each_angle_hook` dispatch/default/legacy-builtins-fallback/unregistered-name,
+`tomo_scan_resume()`/`_resolve_type1_projection()`/`tomo_queue_reacquire()` correctness across all
+three sub-tomogram flavors) and `test_omny_tomo_scan.py` (6 tests -- account handling, progress-GUI
+auto-show, heartbeat clearing on normal completion and mid-scan exception, new-scan timing-field
+resets). `write_pdf_report()`'s `ccm_energy` fallback has no dedicated unit test, matching
+flomni's/lamni's own precedent (neither has one either -- verified live only, real file I/O and
+device reads make it impractical to unit test in isolation). Full suite: 702 passed. Confirmed via
+plain Python (no live client needed) that `OMNY`'s MRO resolves cleanly with `TomoQueueMixin`
+first, no conflicts. **Not yet live-verified against a real running session** (deliberately --
+the local bec-server may be in active use by Mirko's own parallel session; a full live
+`tomo_scan()`/`tomo_queue_execute()` run moves real simulated hardware and consumes real scan
+numbers) -- next step is Mirko pulling and testing live himself, same pattern as this whole
+engagement.
+
 ## Environment note (not code, but will bite again if forgotten)
 
 Both `csaxs_bec` and `bec_widgets` were pip-installed editable pointing at pre-repo-

csaxs_bec/bec_ipython_client/plugins/omny/omny.py

@@ -21,6 +21,7 @@ from csaxs_bec.bec_ipython_client.plugins.omny.omny_sample_transfer_mixin import
     OMNYSampleTransferMixin,
 )
 from csaxs_bec.bec_ipython_client.plugins.omny.x_ray_eye_align import XrayEyeAlign
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin import TomoQueueMixin
 
 logger = bec_logger.logger
 
@@ -54,6 +55,81 @@ class OMNYError(Exception):
 _SUPERUSER_ACCOUNTS: frozenset[str] = frozenset({"gac-x01dc", "holler"})
 
 
+class _ProgressProxy:
+    """Dict-like proxy that persists the OMNY progress dict as a BEC global variable.
+
+    Every read (`proxy["key"]`) fetches the current dict from the global var store,
+    and every write (`proxy["key"] = val`) fetches, updates, and saves it back.
+    This makes the progress state visible to all BEC client sessions via
+    ``client.get_global_var("tomo_progress")`` -- same key and shape as
+    flomni's/lamni's own progress proxy, since only one tomography setup is
+    ever live against a given BEC session. Direct port of
+    ``flomni.py``'s/``lamni.py``'s identical class.
+    """
+
+    _GLOBAL_VAR_KEY = "tomo_progress"
+    _DEFAULTS: dict = {
+        "subtomo": 0,
+        "subtomo_projection": 0,
+        "subtomo_total_projections": 1,
+        "projection": 0,
+        "total_projections": 1,
+        "angle": 0,
+        "tomo_type": 0,
+        "tomo_start_time": None,
+        "estimated_remaining_time": None,
+        "estimated_finish_time": None,
+        "heartbeat": None,
+        "accumulated_idle_time": 0.0,
+    }
+
+    def __init__(self, client):
+        self._client = client
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+    def _load(self) -> dict:
+        val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
+        if val is None:
+            return dict(self._DEFAULTS)
+        return val
+
+    def _save(self, data: dict) -> None:
+        self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
+
+    # ------------------------------------------------------------------
+    # Dict-like interface
+    # ------------------------------------------------------------------
+    def __getitem__(self, key):
+        return self._load()[key]
+
+    def __setitem__(self, key, value) -> None:
+        data = self._load()
+        data[key] = value
+        self._save(data)
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self._load()!r})"
+
+    def get(self, key, default=None):
+        return self._load().get(key, default)
+
+    def update(self, *args, **kwargs) -> None:
+        """Update multiple fields in a single round-trip."""
+        data = self._load()
+        data.update(*args, **kwargs)
+        self._save(data)
+
+    def reset(self) -> None:
+        """Reset all progress fields to their default values."""
+        self._save(dict(self._DEFAULTS))
+
+    def as_dict(self) -> dict:
+        """Return a plain copy of the current progress state."""
+        return self._load()
+
+
 class OMNYInitStagesMixin:
 
     def enable_all_devices(self):
@@ -560,6 +636,7 @@ class OMNYInitStagesMixin:
 
 
 class OMNY(
+    TomoQueueMixin,
     OMNYInitStagesMixin,
     OMNYSampleTransferMixin,
     OMNYAlignmentMixin,
@@ -568,6 +645,73 @@ class OMNY(
     OMNY_rt_client,
     OMNYGuiTools,
 ):
+    # tomo_type values that behave like "type 1" for tomo_queue_reacquire()
+    # (TomoQueueMixin) -- OMNY has three "equally spaced sub-tomograms"
+    # flavors (1/4/5 -> 2/4/8 sub-tomograms), unlike flomni/lamni's single
+    # type-1 flavor (the mixin's default, frozenset({1})).
+    _TOMO_TYPE_1_LIKE: frozenset = frozenset({1, 4, 5})
+
+    # Property names a tomo_queue_add() job snapshots (TomoQueueMixin
+    # contract). Mirrors flomni's own tuple exactly, since OMNY's properties
+    # already mirror flomni's names 1:1 -- no tomo_angle_range/single_point_*/
+    # zero_deg_reference_at_each_subtomo/frames_per_trigger, since OMNY has
+    # no such properties.
+    _TOMO_SCAN_PARAM_NAMES = (
+        "tomo_countingtime",
+        "tomo_shellstep",
+        "fovx",
+        "fovy",
+        "stitch_x",
+        "stitch_y",
+        "tomo_stitch_overlap",
+        "ptycho_reconstruct_foldername",
+        "manual_shift_y",
+        "tomo_type",
+        "tomo_angle_stepsize",
+        "golden_ratio_bunch_size",
+        "golden_max_number_of_projections",
+        "golden_projections_at_0_deg_for_damage_estimation",
+        "corridor_size",
+        "at_each_angle_hook",
+    )
+
+    # Curated allow-list for the "move" queue action -- same shape as
+    # flomni's/lamni's own (TomoQueueMixin contract). "ccm_energy", not
+    # "mokev" -- OMNY has no mokev device in any config (real or
+    # simulated); ccm_energy (bl_optics_hutch.yaml) is always present
+    # regardless of active endstation.
+    _TOMO_QUEUE_MOVE_DEVICES = {
+        "ccm_energy": {"label": "Energy", "unit": "keV"},
+        "idgap": {"label": "Undulator gap", "unit": "mm"},
+    }
+
+    # Same move/optimize_idgap registry as flomni/lamni (TomoQueueMixin
+    # contract) -- optimize_idgap stays a stub (TomoQueueMixin.optimize_idgap
+    # is a no-op), matching both.
+    _TOMO_QUEUE_ACTIONS = {
+        "move": {
+            "func": "_queue_action_move",
+            "idempotent": True,  # absolute moves re-run harmlessly
+            "help": "Move device(s) to absolute position(s).",
+            "params": {"positions": {"type": "device_positions"}},
+        },
+        "optimize_idgap": {
+            "func": "optimize_idgap",
+            "idempotent": True,
+            "help": "Scan idgap over a search range and move to the peak. STUB: no-op.",
+            "params": {
+                "search_range": {
+                    "type": "float",
+                    "default": 0.5,
+                    "min": 0.0,
+                    "max": 2.0,
+                    "unit": "mm",
+                    "label": "Search range",
+                }
+            },
+        },
+    }
+
     def __init__(self, client):
         super().__init__()
         self.client = client
@@ -588,20 +732,69 @@ class OMNY(
         self.corr_angle_y = []
         self.corr_pos_y_2 = []
         self.corr_angle_y_2 = []
-        self.progress = {}
-        self.progress["subtomo"] = 0
-        self.progress["subtomo_projection"] = 0
-        self.progress["subtomo_total_projections"] = 1
-        self.progress["projection"] = 0
-        self.progress["total_projections"] = 1
-        self.progress["angle"] = 0
-        self.progress["tomo_type"] = 0
+        # Progress tracking, persisted via the BEC global variable
+        # "tomo_progress" so it survives a client restart -- deliberately NOT
+        # reset here: this proxy is constructed every time OMNY() is
+        # instantiated (i.e. every new client session), and resetting
+        # unconditionally on every instantiation would wipe tomo_start_time
+        # (and everything else) on exactly the restarts where recovering
+        # scan state matters most. A genuinely new tomo_scan() call resets
+        # the relevant fields itself (see the "new scan" branch in
+        # tomo_scan()).
+        self._progress_proxy = _ProgressProxy(self.client)
         self.OMNYTools = OMNYTools(self.client)
+        self._init_tomo_queue()
         OMNY_rt_client.__init__(self)
         self.align = XrayEyeAlign(self.client, self)
         OMNYGuiTools.__init__(self)
         self.set_client(self.client)
 
+    @property
+    def progress(self) -> _ProgressProxy:
+        """Proxy dict backed by the BEC global variable ``tomo_progress``.
+
+        Readable from any BEC client session via::
+
+            client.get_global_var("tomo_progress")
+
+        Individual fields can be read and written just like a regular dict::
+
+            omny.progress["projection"]          # read
+            omny.progress["projection"] = 42     # write (persists immediately)
+
+        To update multiple fields atomically use :py:meth:`_ProgressProxy.update`::
+
+            omny.progress.update(projection=42, angle=90.0)
+
+        To reset all fields to their defaults::
+
+            omny.progress.reset()
+        """
+        return self._progress_proxy
+
+    @progress.setter
+    def progress(self, val: dict) -> None:
+        """Replace the entire progress dict.
+
+        Accepts a plain :class:`dict` and persists it to the global var store.
+        """
+        if not isinstance(val, dict):
+            raise TypeError(f"progress must be a dict, got {type(val).__name__!r}")
+        self._progress_proxy._save(val)
+
+    def tomo_progress_reset(self) -> None:
+        """Explicitly clear the persisted tomo progress (start time, ETA,
+        current angle/subtomo/projection, accumulated idle time, ...).
+
+        Not called automatically on OMNY() startup -- that would wipe an
+        in-progress scan's state on every kernel restart. Call this by hand
+        if you want a clean progress display without it being tied to
+        starting a new tomo_scan() (which already resets the relevant
+        fields itself).
+        """
+        self._progress_proxy.reset()
+        print("Tomo progress reset.")
+
     def start_x_ray_eye_alignment(self):
         if self.OMNYTools.yesno(
             "Starting Xrayeye alignment. Deleting any potential existing alignment for this sample.",
@@ -1055,6 +1248,8 @@ class OMNY(
 
     def tomo_scan(self, subtomo_start=1, start_angle=None, projection_number=None):
         """start a tomo scan"""
+        self.omnygui_show_progress()
+
         bec = builtins.__dict__.get("bec")
         scans = builtins.__dict__.get("scans")
         self._current_special_angles = self.special_angles.copy()
@@ -1066,11 +1261,13 @@ class OMNY(
         ):
 
             # pylint: disable=undefined-variable
+            # bec.active_account is a plain str, not bytes -- .decode() used
+            # to crash with AttributeError on every real account.
             if bec.active_account != "":
                 self.tomo_id = self.add_sample_database(
                     self.sample_name,
                     str(datetime.date.today()),
-                    bec.active_account.decode(),
+                    bec.active_account,
                     bec.queue.next_scan_number,
                     "OMNY",
                     "test additional info",
@@ -1079,111 +1276,201 @@ class OMNY(
                 self.write_pdf_report()
             else:
                 self.tomo_id = 0
+            self.progress["tomo_start_time"] = datetime.datetime.now().isoformat()
+            # reset stale estimates from any previous scan, otherwise the GUI
+            # would keep showing a leftover ETA from before this scan has
+            # accumulated enough projections to compute a fresh one
+            self.progress["estimated_remaining_time"] = None
+            self.progress["estimated_finish_time"] = None
+            self.progress["accumulated_idle_time"] = 0.0
+            self.progress["heartbeat"] = None
 
-        with scans.dataset_id_on_hold:
-            if self.tomo_type in (1, 4, 5):
-                # equally spaced sub-tomograms (2, 4, or 8)
-                n_subtomos = {1: 2, 4: 4, 5: 8}[self.tomo_type]
-                self.progress["tomo_type"] = f"Equally spaced sub-tomograms ({n_subtomos})"
-                for ii in range(subtomo_start, n_subtomos + 1):
-                    self.sub_tomo_scan(ii, n_subtomos=n_subtomos, start_angle=start_angle)
-                    start_angle = None
+        try:
+            with scans.dataset_id_on_hold:
+                if self.tomo_type in (1, 4, 5):
+                    # equally spaced sub-tomograms (2, 4, or 8)
+                    n_subtomos = {1: 2, 4: 4, 5: 8}[self.tomo_type]
+                    self.progress["tomo_type"] = f"Equally spaced sub-tomograms ({n_subtomos})"
+                    for ii in range(subtomo_start, n_subtomos + 1):
+                        self.sub_tomo_scan(ii, n_subtomos=n_subtomos, start_angle=start_angle)
+                        start_angle = None
 
-            elif self.tomo_type == 2:
-                # Golden ratio tomography
-                previous_subtomo_number = -1
-                if projection_number == None:
-                    ii = 0
-                else:
-                    ii = projection_number
-                while True:
-                    angle, subtomo_number = self._golden(ii, self.golden_ratio_bunch_size, 180, 1)
-                    if previous_subtomo_number != subtomo_number:
-                        self._write_subtomo_to_scilog(subtomo_number)
+                elif self.tomo_type == 2:
+                    # Golden ratio tomography
+                    previous_subtomo_number = -1
+                    if projection_number == None:
+                        ii = 0
+                    else:
+                        ii = projection_number
+                    while True:
+                        angle, subtomo_number = self._golden(
+                            ii, self.golden_ratio_bunch_size, 180, 1
+                        )
+                        if previous_subtomo_number != subtomo_number:
+                            self._write_subtomo_to_scilog(subtomo_number)
+                            if (
+                                subtomo_number % 2 == 1
+                                and ii > 10
+                                and self.golden_projections_at_0_deg_for_damage_estimation == 1
+                            ):
+                                self._tomo_scan_at_angle(0, subtomo_number)
+                            previous_subtomo_number = subtomo_number
+                        self.progress["tomo_type"] = "Golden ratio tomography"
+                        self.progress["subtomo"] = subtomo_number
+                        self.progress["projection"] = ii
+                        self.progress["angle"] = angle
+                        if self.golden_ratio_bunch_size > 0:
+                            self.progress["subtomo_total_projections"] = (
+                                self.golden_ratio_bunch_size
+                            )
+                            self.progress["subtomo_projection"] = (
+                                ii - (subtomo_number - 1) * self.golden_ratio_bunch_size
+                            )
+                        else:
+                            self.progress["subtomo_total_projections"] = 0
+                            self.progress["subtomo_projection"] = 0
+
+                        if self.golden_max_number_of_projections > 0:
+                            self.progress["total_projections"] = (
+                                self.golden_max_number_of_projections
+                            )
+                        else:
+                            self.progress["total_projections"] = 0
+
+                        self._tomo_scan_at_angle(angle, subtomo_number)
+                        ii += 1
                         if (
-                            subtomo_number % 2 == 1
-                            and ii > 10
-                            and self.golden_projections_at_0_deg_for_damage_estimation == 1
+                            ii > self.golden_max_number_of_projections
+                            and self.golden_max_number_of_projections > 0
                         ):
-                            self._tomo_scan_at_angle(0, subtomo_number)
-                        previous_subtomo_number = subtomo_number
-                    self.progress["tomo_type"] = "Golden ratio tomography"
-                    self.progress["subtomo"] = subtomo_number
-                    self.progress["projection"] = ii
-                    self.progress["angle"] = angle
-                    if self.golden_ratio_bunch_size > 0:
-                        self.progress["subtomo_total_projections"] = self.golden_ratio_bunch_size
+                            print(
+                                f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
+                            )
+                            break
+
+                elif self.tomo_type == 3:
+                    # Equally spaced tomography, golden ratio starting angle
+                    previous_subtomo_number = -1
+                    if projection_number == None:
+                        ii = 0
+                    else:
+                        ii = projection_number
+                    while True:
+                        angle, subtomo_number = self._golden_equally_spaced(
+                            ii, int(180 / self.tomo_angle_stepsize), 180, 1, 0
+                        )
+                        if previous_subtomo_number != subtomo_number:
+                            self._write_subtomo_to_scilog(subtomo_number)
+                            if (
+                                subtomo_number % 2 == 1
+                                and ii > 10
+                                and self.golden_projections_at_0_deg_for_damage_estimation == 1
+                            ):
+                                self._tomo_scan_at_angle(0, subtomo_number)
+                            previous_subtomo_number = subtomo_number
+                        self.progress["tomo_type"] = (
+                            "Equally spaced tomography, golden ratio starting angle"
+                        )
+                        self.progress["subtomo"] = subtomo_number
+                        self.progress["projection"] = ii
+                        self.progress["angle"] = angle
+
+                        self.progress["subtomo_total_projections"] = 180 / self.tomo_angle_stepsize
                         self.progress["subtomo_projection"] = (
-                            ii - (subtomo_number - 1) * self.golden_ratio_bunch_size
+                            ii - (subtomo_number - 1) * self.progress["subtomo_total_projections"]
                         )
-                    else:
-                        self.progress["subtomo_total_projections"] = 0
-                        self.progress["subtomo_projection"] = 0
 
-                    if self.golden_max_number_of_projections > 0:
-                        self.progress["total_projections"] = self.golden_max_number_of_projections
-                    else:
-                        self.progress["total_projections"] = 0
-
-                    self._tomo_scan_at_angle(angle, subtomo_number)
-                    ii += 1
-                    if (
-                        ii > self.golden_max_number_of_projections
-                        and self.golden_max_number_of_projections > 0
-                    ):
-                        print(
-                            f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
-                        )
-                        break
-
-            elif self.tomo_type == 3:
-                # Equally spaced tomography, golden ratio starting angle
-                previous_subtomo_number = -1
-                if projection_number == None:
-                    ii = 0
-                else:
-                    ii = projection_number
-                while True:
-                    angle, subtomo_number = self._golden_equally_spaced(
-                        ii, int(180 / self.tomo_angle_stepsize), 180, 1, 0
-                    )
-                    if previous_subtomo_number != subtomo_number:
-                        self._write_subtomo_to_scilog(subtomo_number)
+                        if self.golden_max_number_of_projections > 0:
+                            self.progress["total_projections"] = (
+                                self.golden_max_number_of_projections
+                            )
+                        else:
+                            self.progress["total_projections"] = 0
+                        self._tomo_scan_at_angle(angle, subtomo_number)
+                        ii += 1
                         if (
-                            subtomo_number % 2 == 1
-                            and ii > 10
-                            and self.golden_projections_at_0_deg_for_damage_estimation == 1
+                            ii > self.golden_max_number_of_projections
+                            and self.golden_max_number_of_projections > 0
                         ):
-                            self._tomo_scan_at_angle(0, subtomo_number)
-                        previous_subtomo_number = subtomo_number
-                    self.progress["tomo_type"] = (
-                        "Equally spaced tomography, golden ratio starting angle"
-                    )
-                    self.progress["subtomo"] = subtomo_number
-                    self.progress["projection"] = ii
-                    self.progress["angle"] = angle
+                            print(
+                                f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
+                            )
+                            break
+                else:
+                    raise OMNYError("undefined tomo type")
+        finally:
+            # Cleared on every exit path (normal completion, exception, or
+            # interrupt) -- not just at the next scan's start above -- so the
+            # GUI's busy-detectors see this immediately on their next poll
+            # instead of waiting out the heartbeat-staleness timeout after a
+            # tomogram that's actually already done. Mirrors flomni's/lamni's
+            # identical fix.
+            self.progress["heartbeat"] = None
 
-                    self.progress["subtomo_total_projections"] = 180 / self.tomo_angle_stepsize
-                    self.progress["subtomo_projection"] = (
-                        ii - (subtomo_number - 1) * self.progress["subtomo_total_projections"]
-                    )
+    def tomo_scan_resume(self) -> None:
+        """Resume a tomo_scan() that crashed or was interrupted, picking up
+        automatically from wherever ``progress`` last reported -- no need to
+        read the last subtomo/angle/projection off the progress printout by
+        hand and pass it to tomo_scan() yourself.
 
-                    if self.golden_max_number_of_projections > 0:
-                        self.progress["total_projections"] = self.golden_max_number_of_projections
-                    else:
-                        self.progress["total_projections"] = 0
-                    self._tomo_scan_at_angle(angle, subtomo_number)
-                    ii += 1
-                    if (
-                        ii > self.golden_max_number_of_projections
-                        and self.golden_max_number_of_projections > 0
-                    ):
-                        print(
-                            f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
-                        )
-                        break
-            else:
-                raise OMNYError("undefined tomo type")
+        Re-attempts the exact angle/projection that was in progress at the
+        moment of interruption, rather than skipping ahead to the next one:
+        from here we can't be sure whether that last point was actually
+        acquired before the crash, and re-acquiring one extra projection is
+        harmless.
+
+        Works standalone for a normal tomo_scan(), independent of the tomo
+        queue. tomo_queue_execute() (see TomoQueueMixin) also calls this
+        internally when resuming a job that previously failed partway
+        through, so a failed queue job picks up mid-scan rather than
+        restarting from subtomo 1 / angle 0. Direct port of flomni's/lamni's
+        identical method, generalized to OMNY's three equally-spaced-
+        sub-tomogram flavors (tomo_type 1/4/5) via _TOMO_TYPE_1_LIKE.
+        """
+        if self.progress.get("tomo_start_time") is None:
+            print("No tomo scan in progress to resume -- nothing to do.")
+            return
+
+        if self.tomo_type in self._TOMO_TYPE_1_LIKE:
+            subtomo_start = self.progress["subtomo"]
+            start_angle = self.progress["angle"]
+            if subtomo_start < 1:
+                print("No tomo scan in progress to resume -- nothing to do.")
+                return
+            print(f"Resuming tomo scan at subtomo {subtomo_start}, angle {start_angle:.3f} deg.")
+            self.tomo_scan(subtomo_start=subtomo_start, start_angle=start_angle)
+        elif self.tomo_type in (2, 3):
+            projection_number = self.progress["projection"]
+            print(f"Resuming tomo scan at projection {projection_number}.")
+            self.tomo_scan(projection_number=projection_number)
+        else:
+            raise OMNYError("undefined tomo type")
+
+    def _resolve_type1_projection(self, projection_number: int) -> tuple[int, float]:
+        """Map a flat 0-indexed projection number to (subtomo_start,
+        start_angle), for tomo_queue_reacquire() (TomoQueueMixin) on any
+        tomo_type in _TOMO_TYPE_1_LIKE (1/4/5).
+
+        Reuses _equally_spaced_subtomo_angles() (the single, already-verified
+        source of truth for this angle grid -- see
+        omny/AI_docs/OPEN_ISSUES.md) rather than re-deriving the math here.
+        """
+        n_subtomos = {1: 2, 4: 4, 5: 8}[self.tomo_type]
+        N = int(180.0 / self.tomo_angle_stepsize)
+        total = N * n_subtomos
+        if not 0 <= projection_number < total:
+            raise ValueError(
+                f"projection_number must be in [0, {total}) for tomo_type "
+                f"{self.tomo_type} ({n_subtomos} sub-tomograms, {N} projections each), "
+                f"got {projection_number}."
+            )
+        subtomo_start = projection_number // N + 1
+        projection_within_subtomo = projection_number % N
+        angles = self._equally_spaced_subtomo_angles(
+            subtomo_start, n_subtomos, self.tomo_angle_stepsize
+        )
+        start_angle = float(angles[projection_within_subtomo])
+        return subtomo_start, start_angle
 
     def _print_progress(self):
         print("\x1b[95mProgress report:")
@@ -1210,6 +1497,12 @@ class OMNY(
         return tomo_number
 
     def _at_each_angle(self, angle: float) -> None:
+        hook_name = self.at_each_angle_hook
+        if hook_name:
+            hook = self._resolve_at_each_angle_hook(hook_name, OMNYError)
+            hook(self, angle)
+            return
+
         if "omny_at_each_angle" in builtins.__dict__:
             # pylint: disable=undefined-variable
             omny_at_each_angle(self, angle)
@@ -1555,6 +1848,15 @@ class OMNY(
             fzp_details = self._get_user_param_safe("ofzpx", "fzp_details")
         except Exception:
             fzp_details = "N/A"
+        # The energy readback device is "ccm_energy", not "mokev" -- OMNY has
+        # no mokev device in any config (real or simulated). Read it
+        # defensively so a missing/misbehaving device doesn't crash the
+        # whole report instead of just omitting one line.
+        try:
+            energy_kev = dev.ccm_energy.read(cached=True)[dev.ccm_energy.name]["value"]
+            energy_str = f"{energy_kev:>{padding}.4f}"
+        except Exception:
+            energy_str = f"{'N/A':>{padding}}"
         content = [
             f"{'Sample Name:':<{padding}}{self.sample_name:>{padding}}\n",
             f"{'Measurement ID:':<{padding}}{str(self.tomo_id):>{padding}}\n",
@@ -1564,7 +1866,7 @@ class OMNY(
             f"{'Number of projections:':<{padding}}{int(180 / self.tomo_angle_stepsize * 8):>{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(180 / self.tomo_angle_stepsize * 8) + 10:>{padding}}\n",
-            f"{'Current photon energy:':<{padding}}{dev.mokev.read()['mokev']['value']:>{padding}.4f}\n",
+            f"{'Current photon energy:':<{padding}}{energy_str}\n",
             f"{'Exposure time:':<{padding}}{self.tomo_countingtime:>{padding}.2f}\n",
             f"{'Fermat spiral step size:':<{padding}}{self.tomo_shellstep:>{padding}.2f}\n",
             f"{'FOV:':<{padding}}{fovxy:>{padding}}\n",
@@ -1583,12 +1885,19 @@ class OMNY(
             "xterm /work/sls/spec/local/XOMNY/bin/upload/upload_last_pon.sh &", shell=True
         )
         # status = subprocess.run(f"cp /tmp/spec-e20131-specES1.pdf {user_target}", shell=True)
-        msg = bec.logbook.LogbookMessage()
-        logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "LamNI_logo.png")
-        msg.add_file(logo_path).add_text("".join(content).replace("\n", "</p><p>")).add_tag(
-            ["BEC", "tomo_parameters", f"dataset_id_{dataset_id}", "LamNI", self.sample_name]
-        )
-        self.client.logbook.send_logbook_message(msg)
+        # Same tolerance as write_to_scilog(): a session without scilog/
+        # logbook configured (e.g. a dev/sim session) must not crash report
+        # generation over the logbook upload -- the PDF itself is already
+        # written above.
+        try:
+            msg = bec.logbook.LogbookMessage()
+            logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "OMNY.png")
+            msg.add_file(logo_path).add_text("".join(content).replace("\n", "</p><p>")).add_tag(
+                ["BEC", "tomo_parameters", f"dataset_id_{dataset_id}", "OMNY", self.sample_name]
+            )
+            self.client.logbook.send_logbook_message(msg)
+        except Exception:
+            logger.warning("Failed to write PDF report to scilog.")
 
     def vcs_show_all():
         dev.omny_vcs.show_all()

tests/tests_bec_ipython_client/test_omny_tomo_queue.py

@@ -0,0 +1,317 @@
+"""Unit tests for the tomo-queue / at-each-angle-hook backend ported to
+OMNY from flomni/lamni (see TomoQueueMixin in
+csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py).
+
+Builds an OMNY instance without running __init__'s heavy side effects
+(device init, gui_tools, x-ray-eye alignment, ...) -- only what the
+tomo-queue backend actually needs: self.client + _init_tomo_queue().
+Mirrors test_lamni_tomo_queue.py's structure exactly, plus OMNY-specific
+coverage for its three "equally spaced sub-tomograms" tomo_type flavors
+(1/4/5, via _TOMO_TYPE_1_LIKE) that flomni/lamni don't have.
+"""
+
+import types
+
+import numpy as np
+
+import csaxs_bec.bec_ipython_client.plugins.omny.omny as omny_module
+import csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin as tomo_queue_mixin
+from csaxs_bec.bec_ipython_client.plugins.omny.omny import OMNY, OMNYError, _ProgressProxy
+
+
+class FakeClient:
+    """Minimal in-memory stand-in for BEC's global-var store."""
+
+    def __init__(self):
+        self._vars = {}
+
+    def get_global_var(self, key):
+        return self._vars.get(key)
+
+    def set_global_var(self, key, value):
+        self._vars[key] = value
+
+
+class FakeOMNYTools:
+    def __init__(self, answer=True):
+        self.answer = answer
+
+    def yesno(self, *args, **kwargs):
+        return self.answer
+
+
+def make_omny(fake_client=None):
+    # sample_name reads the module-level `dev` (set at import time inside a
+    # real BEC session only) -- stand in a fake so tomo_queue_execute()'s
+    # confirmation prompt (which reads self.sample_name) doesn't NameError.
+    omny_module.dev = types.SimpleNamespace(
+        omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
+    )
+
+    obj = object.__new__(OMNY)
+    obj.client = fake_client or FakeClient()
+    obj.OMNYTools = FakeOMNYTools()
+    obj._progress_proxy = _ProgressProxy(obj.client)
+    obj._init_tomo_queue()
+    return obj
+
+
+def test_tomo_scan_param_names_have_no_flomni_only_concepts():
+    """OMNY mirrors flomni's names for shared concepts, but has no
+    180/360 toggle, no single-point acquisition, no zero-deg reference,
+    and no frames_per_trigger -- none of these exist as OMNY properties."""
+    names = OMNY._TOMO_SCAN_PARAM_NAMES
+    assert "tomo_angle_range" not in names
+    assert "single_point_instead_of_fermat_scan" not in names
+    assert "single_point_random_shift_max" not in names
+    assert "zero_deg_reference_at_each_subtomo" not in names
+    assert "frames_per_trigger" not in names
+    # shared-with-flomni names are present, using OMNY's own property names
+    assert "fovx" in names
+    assert "stitch_x" in names
+    assert "tomo_angle_stepsize" in names
+
+
+def test_tomo_queue_add_snapshots_all_param_names():
+    omny = make_omny()
+    omny.tomo_countingtime = 0.2
+    omny.fovx = 42.0
+
+    index = omny.tomo_queue_add(label="my job")
+
+    assert index == 0
+    jobs = omny._tomo_queue_proxy.as_list()
+    assert len(jobs) == 1
+    job = jobs[0]
+    assert job["label"] == "my job"
+    assert job["kind"] == "tomo"
+    assert job["status"] == "pending"
+    assert set(job["params"]) == set(OMNY._TOMO_SCAN_PARAM_NAMES)
+    assert job["params"]["tomo_countingtime"] == 0.2
+    assert job["params"]["fovx"] == 42.0
+
+
+def test_tomo_queue_show_survives_missing_param_key(capsys):
+    """A job snapshotted before some key existed in _TOMO_SCAN_PARAM_NAMES
+    won't carry it in its persisted params dict -- tomo_queue_show() must
+    still print a line for it instead of raising KeyError."""
+    omny = make_omny()
+    omny.fovx = 15.0
+    omny.tomo_queue_add(label="job1")
+    jobs = omny._tomo_queue_proxy.as_list()
+    del jobs[0]["params"]["fovx"]
+    omny.client.set_global_var("tomo_queue", jobs)
+
+    result = omny.tomo_queue_show()
+
+    assert result[0]["label"] == "job1"
+    capsys.readouterr()  # just confirm no crash while printing
+
+
+def test_tomo_queue_execute_runs_fresh_job_then_marks_done():
+    omny = make_omny()
+    calls = []
+    omny.tomo_scan = lambda *a, **k: calls.append(("scan", a, k))
+    omny.tomo_scan_resume = lambda *a, **k: calls.append(("resume", a, k))
+
+    omny.tomo_queue_add(label="job1")
+    omny.tomo_queue_execute()
+
+    assert calls == [("scan", (), {"interactive": False})]
+    job = omny._tomo_queue_proxy.as_list()[0]
+    assert job["status"] == "done"
+
+
+def test_tomo_queue_execute_resumes_incomplete_job():
+    omny = make_omny()
+    calls = []
+    omny.tomo_scan = lambda *a, **k: calls.append(("scan", a, k))
+    omny.tomo_scan_resume = lambda *a, **k: calls.append(("resume", a, k))
+
+    index = omny.tomo_queue_add(label="job1")
+    omny._tomo_queue_proxy.update(index, status="incomplete")
+
+    omny.tomo_queue_execute()
+
+    assert calls == [("resume", (), {})]
+    job = omny._tomo_queue_proxy.as_list()[0]
+    assert job["status"] == "done"
+
+
+def test_tomo_queue_add_command_move_action(monkeypatch):
+    omny = make_omny()
+    fake_dev = {"ccm_energy": "ccm-energy-device"}
+    moved = []
+    monkeypatch.setattr(tomo_queue_mixin, "dev", fake_dev, raising=False)
+    monkeypatch.setattr(tomo_queue_mixin, "umv", lambda *a: moved.append(a), raising=False)
+
+    omny.tomo_queue_add_command(
+        {"action": "move", "kwargs": {"positions": {"ccm_energy": 6.2}}}, label="reconfigure"
+    )
+    omny.tomo_queue_execute()
+
+    assert moved == [("ccm-energy-device", 6.2)]
+    job = omny._tomo_queue_proxy.as_list()[0]
+    assert job["kind"] == "command"
+    assert job["status"] == "done"
+
+
+def test_tomo_queue_add_command_rejects_non_movable_device():
+    omny = make_omny()
+    try:
+        omny.tomo_queue_add_command({"action": "move", "kwargs": {"positions": {"fzp": 1.0}}})
+        assert False, "expected ValueError for a non-allow-listed device"
+    except ValueError as exc:
+        assert "not queue-movable" in str(exc)
+
+
+def test_register_at_each_angle_hook_dispatches_from_at_each_angle():
+    omny = make_omny()
+    seen = []
+    omny.register_at_each_angle_hook("record", lambda self, angle: seen.append(angle))
+    omny.at_each_angle_hook = "record"
+
+    omny._at_each_angle(12.5)
+
+    assert seen == [12.5]
+    assert omny.client.get_global_var("tomo_at_each_angle_hooks") == ["record"]
+
+
+def test_at_each_angle_default_path_unchanged_without_hook():
+    """No hook registered -- must fall through to the exact same default
+    behaviour omny had before this port: tomo_scan_projection + tomo_reconstruct."""
+    omny = make_omny()
+    calls = []
+    omny.tomo_scan_projection = lambda angle: calls.append(("projection", angle))
+    omny.tomo_reconstruct = lambda: calls.append(("reconstruct",))
+
+    omny._at_each_angle(30.0)
+
+    assert calls == [("projection", 30.0), ("reconstruct",)]
+
+
+def test_at_each_angle_falls_through_to_legacy_builtins_hook(monkeypatch):
+    """The pre-existing omny_at_each_angle builtins hook must still work as
+    the second-tier fallback, when no at_each_angle_hook is registered."""
+    import builtins
+
+    omny = make_omny()
+    calls = []
+    monkeypatch.setitem(
+        builtins.__dict__, "omny_at_each_angle", lambda setup, angle: calls.append(angle)
+    )
+
+    omny._at_each_angle(7.0)
+
+    assert calls == [7.0]
+
+
+def test_at_each_angle_raises_for_unregistered_hook_name():
+    omny = make_omny()
+    omny.at_each_angle_hook = "missing_hook"
+    try:
+        omny._at_each_angle(0.0)
+        assert False, "expected OMNYError for an unregistered hook name"
+    except OMNYError as exc:
+        assert "missing_hook" in str(exc)
+        assert "not registered" in str(exc)
+
+
+# --- OMNY-specific: tomo_scan_resume() over three sub-tomogram flavors -------
+
+
+def test_tomo_scan_resume_dispatches_subtomo_angle_for_type_1_4_5():
+    for tomo_type, expected_subtomo, expected_angle in [(1, 2, 55.0), (4, 3, 20.0), (5, 7, 90.0)]:
+        omny = make_omny()
+        omny.tomo_type = tomo_type
+        omny.progress["tomo_start_time"] = "2026-01-01T00:00:00"
+        omny.progress["subtomo"] = expected_subtomo
+        omny.progress["angle"] = expected_angle
+        calls = []
+        omny.tomo_scan = lambda *a, **k: calls.append(("scan", a, k))
+
+        omny.tomo_scan_resume()
+
+        assert calls == [
+            ("scan", (), {"subtomo_start": expected_subtomo, "start_angle": expected_angle})
+        ], tomo_type
+
+
+def test_tomo_scan_resume_dispatches_projection_number_for_type_2_3():
+    for tomo_type in (2, 3):
+        omny = make_omny()
+        omny.tomo_type = tomo_type
+        omny.progress["tomo_start_time"] = "2026-01-01T00:00:00"
+        omny.progress["projection"] = 42
+        calls = []
+        omny.tomo_scan = lambda *a, **k: calls.append(("scan", a, k))
+
+        omny.tomo_scan_resume()
+
+        assert calls == [("scan", (), {"projection_number": 42})], tomo_type
+
+
+def test_tomo_scan_resume_noop_when_nothing_in_progress(capsys):
+    omny = make_omny()
+    calls = []
+    omny.tomo_scan = lambda *a, **k: calls.append(a)
+
+    omny.tomo_scan_resume()
+
+    assert calls == []
+    assert "nothing to do" in capsys.readouterr().out
+
+
+# --- OMNY-specific: _resolve_type1_projection() for n_subtomos in {2,4,8} ---
+
+
+def test_resolve_type1_projection_matches_equally_spaced_subtomo_angles():
+    """Cross-check against _equally_spaced_subtomo_angles() itself -- the
+    single source of truth for this angle grid -- rather than
+    re-deriving/duplicating the math in the test."""
+    for tomo_type, n_subtomos in [(1, 2), (4, 4), (5, 8)]:
+        omny = make_omny()
+        omny.tomo_type = tomo_type
+        omny.tomo_angle_stepsize = 10.0
+        N = int(180.0 / omny.tomo_angle_stepsize)
+
+        for projection_number in range(N * n_subtomos):
+            subtomo_start, start_angle = omny._resolve_type1_projection(projection_number)
+            expected_subtomo = projection_number // N + 1
+            expected_angle = OMNY._equally_spaced_subtomo_angles(
+                expected_subtomo, n_subtomos, omny.tomo_angle_stepsize
+            )[projection_number % N]
+            assert subtomo_start == expected_subtomo
+            assert np.isclose(start_angle, expected_angle)
+
+
+def test_resolve_type1_projection_rejects_out_of_range():
+    omny = make_omny()
+    omny.tomo_type = 1
+    omny.tomo_angle_stepsize = 10.0
+    N = int(180.0 / omny.tomo_angle_stepsize)
+    total = N * 2  # n_subtomos=2 for tomo_type 1
+
+    try:
+        omny._resolve_type1_projection(total)
+        assert False, "expected ValueError for an out-of-range projection number"
+    except ValueError:
+        pass
+
+
+def test_tomo_queue_reacquire_dispatches_type1_like_for_types_1_4_5():
+    """tomo_queue_reacquire() (TomoQueueMixin) hardcodes tomo_type == 1 by
+    default; OMNY overrides _TOMO_TYPE_1_LIKE to {1, 4, 5} so reacquiring a
+    type-4/5 job also resolves via _resolve_type1_projection() instead of
+    hitting the mixin's "unknown tomo_type" branch."""
+    for tomo_type, n_subtomos in [(1, 2), (4, 4), (5, 8)]:
+        omny = make_omny()
+        omny.tomo_type = tomo_type
+        omny.tomo_angle_stepsize = 10.0
+        omny.tomo_queue_add(label="job1")
+
+        omny.tomo_queue_reacquire(0, projection_number=1)
+
+        job = omny._tomo_queue_proxy.as_list()[0]
+        assert job["status"] == "incomplete", tomo_type
+        assert omny.progress["subtomo"] in range(1, n_subtomos + 1), tomo_type

tests/tests_bec_ipython_client/test_omny_tomo_scan.py

@@ -0,0 +1,183 @@
+"""Unit tests for tomo_scan()'s new-scan account handling, progress-GUI
+auto-show, and heartbeat-clearing -- all part of porting the tomo-queue
+backend (TomoQueueMixin) to OMNY. Mirrors test_lamni_tomo_angles.py's
+equivalent tests exactly (same bug shapes, same fixes, direct ports of
+flomni's/lamni's already-fixed code).
+
+bec.active_account used to be passed through .decode() (as if it were
+bytes) -- it's a plain str, so this raised AttributeError on every real
+tomo_scan() call with a real account. Unlike flomni/lamni (which always
+attempt sample-database registration, letting it decide the empty-account
+case), OMNY's existing "if bec.active_account != '': ... else: tomo_id = 0"
+branch structure is preserved as-is here -- only the .decode() bug itself
+is fixed, not the branch structure (no TomoIDManager/test-server fallback
+exists in OMNY's own add_sample_database()).
+"""
+
+import builtins
+import types
+
+import pytest
+
+import csaxs_bec.bec_ipython_client.plugins.omny.omny as omny_module
+from csaxs_bec.bec_ipython_client.plugins.omny.omny import OMNY, _ProgressProxy
+
+
+class FakeGlobalVarClient:
+    """Minimal in-memory stand-in for BEC's global-var store."""
+
+    def __init__(self):
+        self._vars = {}
+
+    def get_global_var(self, key):
+        return self._vars.get(key)
+
+    def set_global_var(self, key, value):
+        self._vars[key] = value
+
+
+class _FakeContextManager:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return False
+
+
+class _FakeScans:
+    dataset_id_on_hold = _FakeContextManager()
+
+
+def make_omny_for_tomo_scan(monkeypatch, tomo_angle_stepsize: float, active_account: str) -> OMNY:
+    """Bare OMNY instance with tomo_scan()'s "new scan" branch reachable,
+    everything downstream of it stubbed out (sample database, PDF report,
+    per-sub-tomogram scanning) so only the logic under test actually runs."""
+    # sample_name reads the module-level `dev` (set at import time inside a
+    # real BEC session only) -- stand in a fake so tomo_scan()'s sample-
+    # database registration (which reads self.sample_name) doesn't NameError.
+    omny_module.dev = types.SimpleNamespace(
+        omny_samples=types.SimpleNamespace(get_sample_name_in_samplestage=lambda: "test_sample")
+    )
+
+    obj = object.__new__(OMNY)
+    obj.client = FakeGlobalVarClient()
+    obj._progress_proxy = _ProgressProxy(obj.client)
+    obj.tomo_type = 1
+    obj.tomo_angle_stepsize = tomo_angle_stepsize
+    obj.special_angles = []
+    obj._current_special_angles = []
+    obj.write_pdf_report = lambda: None
+    obj.sub_tomo_scan = lambda subtomo_number, n_subtomos=2, start_angle=None: None
+    obj.omnygui_show_progress = lambda: None
+    monkeypatch.setitem(
+        builtins.__dict__,
+        "bec",
+        types.SimpleNamespace(
+            active_account=active_account, queue=types.SimpleNamespace(next_scan_number=1)
+        ),
+    )
+    monkeypatch.setitem(builtins.__dict__, "scans", _FakeScans())
+    return obj
+
+
+def test_tomo_scan_skips_sample_database_without_active_account(monkeypatch):
+    """Empty active_account (e.g. a dev/sim session) must not crash and must
+    not attempt sample-database registration -- OMNY's own add_sample_database()
+    has no test-server/empty-account fallback, unlike flomni's/lamni's."""
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
+    called = []
+    omny.add_sample_database = lambda *a, **k: called.append(a) or 99
+
+    omny.tomo_scan()
+
+    assert called == []
+    assert omny.tomo_id == 0
+
+
+def test_tomo_scan_registers_sample_with_plain_string_account(monkeypatch):
+    """A real active_account must be passed through as-is -- calling
+    .decode() on it (a plain str, not bytes) raises AttributeError."""
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="e12345")
+    recorded = {}
+
+    def _fake_add_sample_database(samplename, date, eaccount, scan_number, setup, info, user):
+        recorded["eaccount"] = eaccount
+        return 42
+
+    omny.add_sample_database = _fake_add_sample_database
+
+    omny.tomo_scan()
+
+    assert recorded["eaccount"] == "e12345"
+    assert omny.tomo_id == 42
+
+
+def test_tomo_scan_shows_progress_gui(monkeypatch):
+    """tomo_scan() must open/show the progress GUI unconditionally at the
+    start of every scan -- previously only shown per-angle inside
+    _tomo_scan_at_angle(), never at scan start."""
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
+    calls = []
+    omny.omnygui_show_progress = lambda: calls.append("shown")
+
+    omny.tomo_scan()
+
+    assert calls == ["shown"]
+
+
+def test_tomo_scan_clears_heartbeat_on_normal_completion(monkeypatch):
+    """The busy-detector heartbeat must be cleared as soon as the scan
+    finishes, not left for the next poll to time out before the GUI stops
+    showing "beamline busy" for an already-finished scan.
+
+    sub_tomo_scan() is stubbed to set a fresh heartbeat before returning,
+    the same way the real _tomo_scan_at_angle() does at the start of every
+    projection -- without that, this test would pass even without the fix,
+    since tomo_scan()'s own "new scan" branch already resets the heartbeat
+    to None *before* the scan body runs.
+    """
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
+
+    def _fake_sub_tomo_scan(subtomo_number, n_subtomos=2, start_angle=None):
+        omny.progress["heartbeat"] = "2026-01-01T00:00:00"
+
+    omny.sub_tomo_scan = _fake_sub_tomo_scan
+
+    omny.tomo_scan()
+
+    assert omny.progress["heartbeat"] is None
+
+
+def test_tomo_scan_clears_heartbeat_even_if_scan_raises(monkeypatch):
+    """The heartbeat must be cleared on every exit path, including a
+    mid-scan exception -- otherwise a crashed scan looks permanently "busy"
+    to the GUI until the staleness timeout, or forever if polled more often
+    than that."""
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
+
+    def _boom(subtomo_number, n_subtomos=2, start_angle=None):
+        omny.progress["heartbeat"] = "2026-01-01T00:00:00"
+        raise RuntimeError("simulated crash mid-scan")
+
+    omny.sub_tomo_scan = _boom
+
+    with pytest.raises(RuntimeError):
+        omny.tomo_scan()
+
+    assert omny.progress["heartbeat"] is None
+
+
+def test_tomo_scan_sets_timing_fields_on_new_scan(monkeypatch):
+    """A fresh tomo_scan() must stamp tomo_start_time and reset stale
+    ETA/idle-time estimates from any previous scan."""
+    omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
+    omny.progress["estimated_remaining_time"] = "stale"
+    omny.progress["estimated_finish_time"] = "stale"
+    omny.progress["accumulated_idle_time"] = 123.0
+
+    omny.tomo_scan()
+
+    assert omny.progress["tomo_start_time"] is not None
+    assert omny.progress["estimated_remaining_time"] is None
+    assert omny.progress["estimated_finish_time"] is None
+    assert omny.progress["accumulated_idle_time"] == 0.0