first iteration

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -5,6 +5,7 @@ import os
 import random
 import subprocess
 import time
+import uuid
 from pathlib import Path
 
 import h5py
@@ -1659,6 +1660,41 @@ class _TomoQueueProxy:
         jobs[index].update(kwargs)
         self._save(jobs)
 
+    def update_by_id(self, job_id: str, **kwargs) -> bool:
+        """Update the job whose ``id`` matches ``job_id``.
+
+        Returns True if a matching job was found and updated, False otherwise
+        (e.g. the job was deleted concurrently). Addressing a job by its stable
+        id rather than its list index makes status writes safe against queue
+        edits made while the queue runs: a reorder, or a delete of some *other*
+        job, can no longer misdirect a status write to the wrong entry.
+        """
+        jobs = self._load()
+        for job in jobs:
+            if job.get("id") == job_id:
+                job.update(kwargs)
+                self._save(jobs)
+                return True
+        return False
+
+    def ensure_ids(self) -> list:
+        """Assign a stable id to any job that lacks one and persist the result.
+
+        Back-compat shim for queues written before jobs carried an ``id`` field
+        (and for any job added by older code). Called once at the top of
+        tomo_queue_execute() so the pick-next executor can address every job by
+        id. Returns the (possibly healed) job list.
+        """
+        jobs = self._load()
+        changed = False
+        for job in jobs:
+            if not job.get("id"):
+                job["id"] = uuid.uuid4().hex
+                changed = True
+        if changed:
+            self._save(jobs)
+        return jobs
+
     def clear(self) -> None:
         self._save([])
 
@@ -3559,6 +3595,7 @@ class Flomni(
         params = {name: getattr(self, name) for name in self._TOMO_QUEUE_PARAM_NAMES}
         index = len(self._tomo_queue_proxy)
         job = {
+            "id": uuid.uuid4().hex,
             "label": label or f"job_{index + 1}",
             "params": params,
             "status": "pending",
@@ -3613,9 +3650,29 @@ class Flomni(
         ``tomo_scan()`` -- or, for a job that didn't run to completion last
         time, ``tomo_scan_resume()``, so it picks back up mid-scan instead
         of restarting from subtomo 1 / angle 0. Jobs already marked "done"
-        are skipped on the next call, so simply calling tomo_queue_execute()
-        again resumes from where it stopped (e.g. after fixing a hardware
-        issue).
+        are skipped, so simply calling tomo_queue_execute() again resumes
+        from where it stopped (e.g. after fixing a hardware issue).
+
+        Pick-next execution model
+        -------------------------
+        The queue is re-read from its global var at the start of every job
+        (not snapshotted once up front), and each job is addressed by its
+        stable ``id``, not by its list index. This is what makes it safe to
+        edit the queue *while it runs*: appending a job, deleting a pending
+        job, or reordering the pending tail all take effect at the next job
+        boundary, and a reorder/delete can never misdirect a status write to
+        the wrong entry (an index-based loop could). Legacy jobs written
+        before the ``id`` field existed are healed once up front by
+        ensure_ids().
+
+        Resume-before-fresh ordering: if any job is in a resumable state, it
+        is always run before any fresh "pending" job, regardless of list
+        order. This preserves the single-resumable-job invariant -- tomo
+        progress lives in one shared ``progress`` global var, so starting a
+        fresh tomo_scan() before resuming an interrupted one would overwrite
+        the very state tomo_scan_resume() needs. (Under the old strictly
+        in-order loop this couldn't happen; allowing reorder makes it
+        possible, hence the explicit guard.)
 
         A job is considered not-yet-complete (and so gets resumed rather
         than restarted) if its status is "incomplete" (a Python exception
@@ -3632,51 +3689,85 @@ class Flomni(
         yourself to recover from a crash (bypassing the queue), that scan
         is now actually finished even though the queue still has the job
         marked "incomplete" or "running" -- mark it done yourself before
-        calling this again, or it will be re-run from scratch:
-            flomni._tomo_queue_proxy.update(job_index, status="done")
+        calling this again, or it will be re-run from scratch. Address the
+        job by its id (indices can shift if the queue is reordered):
+            flomni._tomo_queue_proxy.update_by_id(job_id, status="done")
 
         Args:
-            start_index: Queue index to start from. Defaults to 0, but jobs
-                already marked "done" are skipped automatically either way.
+            start_index: Position in the *current* queue order below which
+                fresh "pending" jobs are ignored. Defaults to 0. Resumable
+                jobs are always run regardless of start_index (a crashed
+                job must be recovered wherever it sits). Jobs already marked
+                "done" are skipped automatically either way. Note that with
+                reordering active this is applied to the queue's current
+                order at each pick.
         """
+        self._tomo_queue_proxy.ensure_ids()
         jobs = self._tomo_queue_proxy.as_list()
         if not jobs:
             print("Tomo queue is empty.")
             return
 
+        runnable = [
+            job
+            for idx, job in enumerate(jobs)
+            if job["status"] != "done"
+            and (job["status"] in ("incomplete", "running") or idx >= start_index)
+        ]
+        if not runnable:
+            print("No pending tomo queue jobs to run.")
+            return
+
         if not self.OMNYTools.yesno(
-            f"Starting automatic execution of {len(jobs) - start_index} queued tomo scan(s) on"
+            f"Starting automatic execution of {len(runnable)} queued tomo scan(s) on"
             f" sample '{self.sample_name}'. OK?",
             "y",
         ):
             print("Aborted.")
             return
 
-        for i in range(start_index, len(jobs)):
-            job = jobs[i]
-            if job["status"] == "done":
-                continue
+        while True:
+            jobs = self._tomo_queue_proxy.as_list()
+            # Resume any in-flight/interrupted job first, before starting any
+            # fresh one, so the shared progress var is never clobbered. Search
+            # the whole list (not the start_index slice): a crashed job must be
+            # recovered wherever it sits.
+            job = next((j for j in jobs if j["status"] in ("incomplete", "running")), None)
+            if job is None:
+                job = next(
+                    (
+                        j
+                        for idx, j in enumerate(jobs)
+                        if idx >= start_index and j["status"] == "pending"
+                    ),
+                    None,
+                )
+            if job is None:
+                break
+
+            job_id = job["id"]
+            label = job["label"]
             resume_job = job["status"] in ("incomplete", "running")
 
-            print(f"\n=== Tomo queue job {i + 1}/{len(jobs)}: {job['label']} ===")
+            print(f"\n=== Tomo queue job: {label} ===")
             for name, value in job["params"].items():
                 setattr(self, name, value)
 
-            self._tomo_queue_proxy.update(i, status="running")
+            self._tomo_queue_proxy.update_by_id(job_id, status="running")
             try:
                 if resume_job:
                     self.tomo_scan_resume()
                 else:
                     self.tomo_scan()
             except Exception as exc:
-                self._tomo_queue_proxy.update(i, status="incomplete")
-                print(f"Tomo queue job {i} ({job['label']}) did not complete: {exc}")
+                self._tomo_queue_proxy.update_by_id(job_id, status="incomplete")
+                print(f"Tomo queue job '{label}' did not complete: {exc}")
                 print(
                     "Queue paused. Fix the issue and call tomo_queue_execute() "
                     "again to resume from this job."
                 )
                 raise
-            self._tomo_queue_proxy.update(i, status="done")
+            self._tomo_queue_proxy.update_by_id(job_id, status="done")
 
         print("\nTomo queue finished -- all jobs done.")
 

csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml

@@ -180,7 +180,7 @@ fsamroy:
     - 365
     port: 8084
     sign: -1
-    sim_velocity: 25
+    sim_velocity: 250 # 10x hardware
     sim_initial_position: 0
   enabled: true
   onFailure: buffer

pyproject.toml

@@ -38,6 +38,7 @@ dev = [
     "pylint",
     "pytest",
     "pytest-random-order",
+    "pytest-bec-e2e",
     "ophyd_devices",
     "bec_server",
 ]