From ff50711dc9738b3a5ae2df4bbd478173d85820b3 Mon Sep 17 00:00:00 2001
From: Mirko Holler <holler@Mac.fritz.box>
Date: Sun, 12 Jul 2026 23:25:33 +0200
Subject: [PATCH] test(flomni): add tomo-queue e2e harness against the
 simulated flOMNI

Sections A-C of TOMO_QUEUE_TESTING.md's checklist, run against a live sim
session: params-restored-per-job, legacy queue migration, empty/all-done
no-ops, start_index semantics, exception/SIGKILL crash-resume, resume-
before-fresh ordering, and concurrent queue edits from a second client.
All 10 tests pass together with no cross-test interference.

The flomni_sim fixture and _bootstrap.py factor out what it takes to
construct a live Flomni against the sim (builtins/reload bootstrap,
side-effect neutralization, RT-feedback/fsamx setup) for reuse by both
in-process tests and the SIGKILL-based subprocess tests.

Also adds the AI_docs/ handoff docs (testing checklist results, command-
jobs plan with the action-registry decisions) that this work updated.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 .../AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md   | 510 ++++++++++++++++++
 .../flomni/AI_docs/TOMO_QUEUE_TESTING.md      | 352 ++++++++++++
 tests/e2e/_bootstrap.py                       | 113 ++++
 tests/e2e/_queue_helpers.py                   | 120 +++++
 tests/e2e/_run_queue_subprocess.py            |  36 ++
 tests/e2e/conftest.py                         |  37 ++
 tests/e2e/test_smoke.py                       |   8 +
 tests/e2e/test_tomo_queue.py                  | 159 ++++++
 tests/e2e/test_tomo_queue_concurrency.py      | 120 +++++
 tests/e2e/test_tomo_queue_crash.py            |  89 +++
 tests/e2e/test_tomo_queue_crash_subprocess.py |  89 +++
 11 files changed, 1633 insertions(+)
 create mode 100644 csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md
 create mode 100644 csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_TESTING.md
 create mode 100644 tests/e2e/_bootstrap.py
 create mode 100644 tests/e2e/_queue_helpers.py
 create mode 100644 tests/e2e/_run_queue_subprocess.py
 create mode 100644 tests/e2e/conftest.py
 create mode 100644 tests/e2e/test_smoke.py
 create mode 100644 tests/e2e/test_tomo_queue.py
 create mode 100644 tests/e2e/test_tomo_queue_concurrency.py
 create mode 100644 tests/e2e/test_tomo_queue_crash.py
 create mode 100644 tests/e2e/test_tomo_queue_crash_subprocess.py

diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md
new file mode 100644
index 0000000..745120c
--- /dev/null
+++ b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md
@@ -0,0 +1,510 @@
+# Plan: Command Jobs in the FlOMNI Tomo Queue
+
+**Status:** design agreed, **and the §10 open questions are now resolved with Mirko**
+(one small detail — `optimize_idgap`'s peak-detection signal — deferred to the
+implementation session). **Step 1 (executor rework) is implemented and its sim
+checklist has passed** (`TOMO_QUEUE_TESTING.md` §4, sections A–C). Steps 2 and 3
+below are not yet written — this is now unblocked to start. Implement in a dedicated
+session — this touches the CLI (`flomni.py`), the GUI widget (`tomo_params.py` /
+`TomoQueueDialog`), and the web page generator (`flomni_webpage_generator.py`).
+
+**Prerequisite (met):** the checklist in `TOMO_QUEUE_TESTING.md` has passed against
+the simulated flOMNI. That document is the companion to this one: it covers *what
+Step 1 changed and how to test it*; this one covers *what comes next*.
+
+> **This document supersedes the earlier version of the same name.** The earlier
+> draft assumed the old index-based executor, single-action command jobs, and left
+> the registry-distribution question open. All three have since been decided.
+
+**Goal:** Allow the tomo queue to hold, in addition to tomogram jobs, "command" jobs
+that reconfigure the beamline between tomograms — move a device, change energy, move
+optics in/out, optimize a parameter — so one queue can express *"tomogram A, then
+reconfigure, then tomogram B"*, unattended.
+
+---
+
+## 0. Chosen model, and what was rejected
+
+**Chosen: a named-action registry.** Command jobs name actions from a curated
+allow-list, plus keyword arguments. Everything stays JSON-serializable, auditable in
+`client.get_global_var("tomo_queue")`, and safe to persist across kernel restarts.
+
+**Rejected: storing Python source strings and `exec`-ing them later.** Not auditable;
+fails at run time (hours into an overnight run) rather than at add time; breaks the
+crash-resume model. If the registry ever can't express something, adding an entry is a
+small, reviewed change — preferable to a permanent arbitrary-code channel on a live
+beamline.
+
+**Also rejected: a code editor in the GUI with a syntax check.** Syntax-valid is nearly
+worthless as a safety check — `umv(dev.fsamx, -1e9)` parses perfectly and drives the
+stage into a hard stop. A syntax check gives the operator *false confidence*. The
+structured builder (§4) is both safer and friendlier: there is nothing to type wrong.
+
+---
+
+## 1. Current state
+
+### Step 1 — DONE (executor rework)
+
+`tomo_queue_execute()` is now an **id-based pick-next loop**: it re-reads the queue
+from its global var at the start of every job and addresses jobs by a stable `id`
+(not by list index). `_TomoQueueProxy` gained `update_by_id()` and `ensure_ids()`;
+`tomo_queue_add()` stamps a uuid on each job.
+
+This is what makes it safe to **append, delete a pending job, and reorder the pending
+tail while the queue runs** — the foundation the GUI work (§5) depends on. It also
+introduced **resume-before-fresh** ordering: any `incomplete`/`running` job runs before
+any fresh `pending` one, protecting the single shared `progress` global var.
+
+See `TOMO_QUEUE_TESTING.md` §1 for the full rationale and the invariants to preserve.
+
+### Job dict schema today
+
+```python
+{
+    "id": str,                      # uuid4 hex; stable across reorder/edit
+    "label": str,
+    "params": {<_TOMO_QUEUE_PARAM_NAMES>: value, ...},
+    "status": "pending" | "running" | "incomplete" | "done",
+    "added_at": ISO datetime string,
+}
+```
+
+Every job today is *implicitly* a tomogram. Adding command jobs = making the job
+**type-tagged** and branching in the executor.
+
+### The GUI as it stands
+
+`TomoQueueDialog` (in `tomo_params.py`) reads/writes the same `tomo_queue` global var
+directly, renders one row per job, colour-codes by status, and polls every 2 s.
+Execution is CLI-only (the GUI server process can't run the blocking scan loop).
+
+---
+
+## 2. Schema: type-tagged jobs
+
+Add a `"kind"` field. **Backward compatible:** a job with no `kind` is treated as
+`"tomo"`, so existing persisted queues and existing GUI code keep working.
+
+### Tomo job (unchanged except the tag)
+```python
+{
+    "kind": "tomo",                 # NEW; absent => "tomo"
+    "id": str,
+    "label": str,
+    "params": {...},
+    "status": ...,
+    "added_at": ...,
+}
+```
+
+### Command job (new) — a **list of steps**, not a single action
+```python
+{
+    "kind": "command",
+    "id": str,
+    "label": str,
+    "steps": [                      # ordered; run in sequence within the one job
+        {"action": "move", "kwargs": {"positions": {"mokev": 6.2}}},
+        {"action": "optimize_idgap", "kwargs": {"search_range": 0.5}},
+    ],
+    "idempotent": bool,             # true iff EVERY step is idempotent
+    "status": ...,
+    "added_at": ...,
+}
+```
+(`foptics_out`/`foptics_in`-style steps aren't in the initial registry — see §3 —
+this example only uses the two actions Step 2 actually ships with.)
+
+**Why a list.** "Between tomogram A and B: change energy, re-peak idgap" is one
+*intent* and should be one queue row — it reads correctly, and it is atomic to reason
+about. Composing it from three separate queue rows is worse on both counts.
+
+Notes:
+- `kwargs` must be JSON-serializable. No device objects, no callables — actions resolve
+  device names to `dev.*` themselves at execution time.
+- Steps within a job are **independent**: none of them consumes another's runtime
+  result. Anything result-dependent belongs *inside* a single action (§3.2).
+
+---
+
+## 3. The action registry
+
+A class-level dict on `Flomni` mapping an action name to a spec. Callables are existing
+`Flomni`/mixin methods or small reviewed wrappers, so the queue can only ever invoke
+reviewed beamline operations.
+
+**Each registry entry has four things:**
+
+| field | meaning |
+|---|---|
+| `func` | name of the bound method to call |
+| `idempotent` | default: safe to re-run after a crash? |
+| `help` | short operator-facing description |
+| `params` | **schema** of the action's arguments (possibly empty) |
+
+**Decided with Mirko: the initial registry is deliberately minimal — `move` and
+`optimize_idgap` only.** No `foptics_in`/`foptics_out`/`feye_in`/`feye_out`/shutter
+action for now, even though the first three already exist as callable methods —
+Step 2 ships with move-on-allow-listed-devices plus one compound action; further
+one-click actions are "dedicated scripts" to be written and added later as their own
+reviewed changes, not part of this rollout. No dedicated energy-change wrapper either
+— energy changes go through `move` on `mokev` (see §3.1), not a special action.
+
+```python
+_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"},   # {device: target}, allow-listed
+        },
+    },
+    # compound actions — see §3.2
+    "optimize_idgap": {
+        "func": "optimize_idgap",
+        "idempotent": True,
+        "help": "Scan idgap over a search range and move to the peak.",
+        "params": {
+            "search_range": {
+                "type": "float", "default": 0.5, "min": 0.0, "max": 2.0,
+                "unit": "mm", "label": "Search range",
+            },
+        },
+    },
+}
+```
+
+**Still open (confirm before/while implementing `optimize_idgap`):** which
+signal/device the scan reads to locate "the peak" — deferred by Mirko to the
+implementation session, not resolved yet. The `search_range` default/bounds above
+are confirmed as "roughly right" placeholders; unit (`mm`) not explicitly
+reconfirmed but not contradicted either.
+
+### 3.1 Curated device allow-list (NOT "any device in `dev`")
+
+The `move` action may only touch devices on an explicit, reviewed allow-list — the
+things you actually reconfigure between tomograms. "Any device in `dev`" is the
+difference between *a reconfiguration tool* and *a remote control for the whole
+endstation*.
+
+**Decided with Mirko: `mokev` (energy) and `idgap` (undulator gap) — nothing else
+yet.** Neither device is in the flomni sim config (`simulated_flomni.yaml`); both
+live in `bl_frontend.yaml` (shared beamline front-end devices), so `move` on either
+is only exercisable against the real front end, not the flomni sim used for
+`TOMO_QUEUE_TESTING.md`.
+
+```python
+_TOMO_QUEUE_MOVE_DEVICES = {
+    "mokev": {"label": "Energy", "unit": "keV"},
+    "idgap": {"label": "Undulator gap", "unit": "mm"},   # unit unconfirmed, see above
+}
+
+def _queue_action_move(self, positions: dict):
+    """positions: {device_name: absolute_target, ...}. Absolute umv only."""
+    args = []
+    for name, target in positions.items():
+        if name not in self._TOMO_QUEUE_MOVE_DEVICES:
+            raise ValueError(f"Queue move: device '{name}' is not queue-movable.")
+        if name not in dev:
+            raise ValueError(f"Queue move: unknown device '{name}'.")
+        args += [dev[name], target]
+    umv(*args)
+```
+
+The GUI's device dropdown is built from this allow-list, so an operator cannot even
+select a device that isn't sanctioned.
+
+### 3.2 Compound actions, and where result-dependent logic lives
+
+**Rule: anything where step B needs step A's *runtime result* is ONE named action, not
+several queue steps.**
+
+The motivating case is "scan idgap, move to peak". That is not three primitives — it is
+one compound operation with an internal result. It becomes a single registry entry
+(`optimize_idgap`) whose method does the scan, finds the peak, and moves there. The
+cleverness lives in reviewed Python; the queue only ever holds the *intent*:
+
+```python
+{"action": "optimize_idgap", "kwargs": {"search_range": 0.5}}
+```
+
+Why this matters: the queue then **never has to carry runtime values between steps**, so
+`kwargs` stays plain JSON and crash-resume stays honest. Trying to express "move to
+wherever the peak turned out to be" as separate queue steps is exactly what would force
+a code channel.
+
+**Compound actions may take parameters** (`search_range` above) — declared in the same
+`params` schema as any other action. Parameters are *scalar inputs the operator sets at
+add time*: numbers, bools, a choice from a fixed list, a device from the allow-list. The
+schema is deliberately **not** expressive enough to encode logic, expressions, or
+references to other jobs' results. That limitation is the point — a parameter is a value
+you can validate at add time.
+
+### 3.3 Validation is two-layer
+
+- **Schema validation at add time** (GUI + `tomo_queue_add_command`): types, min/max,
+  device on the allow-list, OK disabled until valid. This is for *ergonomics* — catch
+  the typo now, not at 2am.
+- **Method validation at execution time**: each action re-checks its own arguments
+  against live hardware. This is the *safety boundary*, because the schema and the
+  hardware can disagree (limits changed; a value arrived via the CLI bypassing the
+  form). A refusal here trips the executor's `incomplete`-and-pause path, which is the
+  correct outcome.
+
+---
+
+## 4. New / changed CLI methods (`flomni.py`)
+
+### `tomo_queue_add_command(steps, label=None, idempotent=None)`
+
+Validate every step's `action` against `_TOMO_QUEUE_ACTIONS`; validate its `kwargs`
+against the action's `params` schema (including the device allow-list for `move`);
+confirm JSON-serializability (round-trip `json.dumps`); resolve `idempotent`
+(explicit arg > **all steps idempotent**); append a `kind="command"` job with a fresh
+uuid. Return the new index. Mirror the print style of `tomo_queue_add`.
+
+Convenience for the common single-step case is fine (`steps` may accept a single dict),
+but the stored schema is always a list.
+
+```python
+flomni.tomo_queue_add_command(
+    [{"action": "move", "kwargs": {"positions": {"mokev": 6.2}}},
+     {"action": "optimize_idgap", "kwargs": {"search_range": 0.5}}],
+    label="reconfigure to 6.2 keV",
+)
+```
+
+### `tomo_queue_add(...)` — unchanged behaviour
+Keep as-is, but write `"kind": "tomo"` explicitly going forward (new jobs
+self-describing; old ones without the field still default to tomo on read).
+
+### `tomo_queue_show()` — render both kinds
+Branch on `kind`. Command rows: show the step sequence compactly + an idempotency
+marker. Keep column alignment so the two kinds read as one list.
+
+```
+[2]    pending  reconfigure 6.2keV   CMD move{mokev:6.2} > optimize_idgap{search_range:0.5}  [idem]
+```
+
+### `tomo_queue_execute()` — branch on kind
+Inside the existing **pick-next loop** (do not reintroduce an index loop), after
+selecting the job and determining `resume_job`:
+
+```python
+kind = job.get("kind", "tomo")
+self._tomo_queue_proxy.update_by_id(job_id, status="running")
+try:
+    if kind == "command":
+        self._run_command_job(job, resume_job)
+    else:  # "tomo"
+        for name, value in job["params"].items():
+            setattr(self, name, value)
+        if resume_job:
+            self.tomo_scan_resume()
+        else:
+            self.tomo_scan()
+except Exception as exc:
+    self._tomo_queue_proxy.update_by_id(job_id, status="incomplete")
+    ...  # existing pause/hint/raise
+self._tomo_queue_proxy.update_by_id(job_id, status="done")
+```
+
+Note the tomo branch reads `job["params"]` — the command branch must never touch it
+(command jobs have no `params` key).
+
+### `_run_command_job(job, resume_job)`
+Apply the resume policy (§5), then run each step in order: resolve the action spec,
+`getattr(self, spec["func"])`, call with `**step["kwargs"]`.
+
+---
+
+## 5. Crash-resume semantics
+
+Tomo jobs resume mid-scan via `tomo_scan_resume()` because progress lives in global
+vars. Command jobs have no such state. Policy:
+
+**No per-step resume cursor.** Deliberately: tracking which step of a multi-step job
+completed means more persisted state and more edge cases than this warrants.
+
+**The job-level `idempotent` flag decides:**
+
+- `idempotent == True` (all steps idempotent — absolute moves, in/out, a scan-to-peak):
+  silently **re-run the whole step sequence from the top**. Re-running an absolute move
+  or an in-out is harmless.
+- `idempotent == False` (any relative/stateful step): **do not silently re-run.** Stop
+  and prompt the operator (`OMNYTools.yesno`): *"Command job '<label>' may have
+  partially run before the crash. Re-run it? (y/n)"*. On "no", mark it `done` and
+  continue; on "yes", re-run from the top.
+
+Edge case handled transparently: a command job that *completed* but crashed before its
+status hit `done` (stuck at `running`). Idempotent → re-runs harmlessly. Non-idempotent
+→ asks, which is the safe default.
+
+**Write actions to be idempotent wherever possible.** Anything relative or stateful
+(e.g. `bec.queue.next_dataset_number += 1`, feedback toggles) must be marked
+`idempotent=False` if ever added.
+
+---
+
+## 6. GUI (`tomo_params.py` / `TomoQueueDialog`)
+
+The agreed interaction model. Its whole purpose: **the GUI stays usable while a
+tomogram is running** — you can queue more work without touching the running scan.
+
+### 6.1 The load-bearing rule: two separate write paths
+
+| control | writes | allowed while a scan runs? |
+|---|---|---|
+| **Done** | the **live param global vars** | **NO — must be blocked** |
+| **Add to queue** (new) | **only** the `tomo_queue` var | **yes, always** |
+
+This separation is the entire reason the GUI can be used during a run. The scan
+**re-reads the live param global vars on every projection** (`fovx`, `fovy`,
+`tomo_shellstep`, `tomo_countingtime`, `frames_per_trigger`, `stitch_x/y`,
+`corridor_size`, …), so writing them mid-scan *perturbs the running acquisition*.
+"Add to queue" must never touch them.
+
+**"Add to queue" cannot call the CLI `tomo_queue_add()`** — that snapshots the *live*
+vars and would ignore the operator's unapplied form edits. The dialog must assemble the
+job dict itself: edited form fields, plus current live values for the params the form
+doesn't expose, so the snapshot is still a complete `_TOMO_QUEUE_PARAM_NAMES` set.
+
+Gate "Done" on the same busy signal `TomoParamsWidget` already computes (tomo heartbeat
+OR BEC scan-queue status).
+
+### 6.2 No in-place row editing
+
+Editing a queued job = **delete it, add a new one, reorder into place**. One
+construction path for jobs; no snapshot-vs-live ambiguity for existing rows.
+
+### 6.3 Row locks keyed on `status` (not on the global busy signal)
+
+Use the *row's own* status, because a bare `flomni.tomo_scan()` leaves no queue row
+marked `running` while a queue run marks exactly one — so per-row status is the precise
+"this is the live item" signal.
+
+- `running`: **not draggable, not deletable.** Deletable only after the measurement is
+  stopped (it is then `incomplete`).
+- `pending` / `incomplete`: freely draggable and deletable.
+- `done`: pinned (harmless to move, but it makes the list confusing to read).
+
+### 6.4 Reorder is allowed **mid-run**
+
+It takes effect at the next job boundary: the executor finishes the running job, marks
+it `done`, re-reads the queue, and picks the first `pending` in the *current* order. No
+interruption, no index math.
+
+**Reject any drop that would move a row above the running row.** The running row is a
+hard floor for drops — otherwise the list would read as if a pending job runs "before"
+the thing already in progress.
+
+### 6.5 The command-job builder — a structured form, not a text box
+
+An **"Add command…"** control opens a sub-dialog:
+
+- an ordered **step list** (add / remove / reorder steps before confirming),
+- per step: an **action dropdown** populated from the registry, then **only the fields
+  that action declares** in its `params` schema — for `move`, a device dropdown (from
+  the allow-list) + numeric target; for `optimize_idgap`, a `search_range` field
+  pre-filled with its default; for parameterless actions, nothing,
+- numeric fields validated live (schema `min`/`max`, and the device's real
+  `dev.<name>.limits` where applicable) — **OK disabled until every step is valid**,
+- a label field and an `idempotent` checkbox pre-filled from the steps' registry
+  defaults.
+
+On confirm, append a `kind="command"` job in exactly the format the CLI writes, so CLI
+and GUI stay interchangeable.
+
+### 6.6 Unchanged
+
+Execution stays **CLI-only** (`flomni.tomo_queue_execute()`); the dialog keeps its
+"execute from CLI" hint. The 2 s poll timer and `closeEvent`/`showEvent` logic are
+untouched. Keep the existing `fsamroy` setup guard — command actions are
+FlOMNI-specific; don't expose command-add in non-FlOMNI sessions.
+
+---
+
+## 7. Registry distribution: publish via global var
+
+On `Flomni.__init__`, write a serializable description of `_TOMO_QUEUE_ACTIONS` +
+`_TOMO_QUEUE_MOVE_DEVICES` (names, params schemas, idempotent defaults, help) to a
+`tomo_queue_actions` global var. The GUI reads it and builds its dropdowns and forms
+from it.
+
+**Decided: this, not a static mirror.** A hardcoded mirror in the widget would drift —
+exactly the problem `QUEUE_PARAM_NAMES` already has between `flomni.py` and
+`tomo_params.py`. With the global var, a new action is a registry entry plus the method;
+the GUI grows the right fields with **zero widget changes**.
+
+---
+
+## 8. Web page generator (`flomni_webpage_generator.py`)
+
+Confirm exactly what it renders for the queue/progress when implementing. It must not
+break on a job with `kind="command"` and **no `params` key**:
+
+- Any code doing `job["params"][...]` will `KeyError` — guard with
+  `job.get("kind", "tomo") == "tomo"` before touching `params`.
+- Add a sensible rendering for command jobs (steps summary + status), consistent with
+  how tomo jobs are shown.
+- Re-check any "projections remaining" / ETA computation so command jobs (which have no
+  projections) don't skew tomogram-based estimates.
+
+---
+
+## 9. Validation & delivery
+
+Per house rules (see `TOMO_QUEUE_TESTING.md` §0): `ast.parse` + no **new** `black`
+hunks (`--line-length=100 --skip-magic-trailing-comma`) on every edited file; minimal
+diffs; Conventional Commits; **sim validation is the gate**.
+
+Behavioural checks (mock first, then sim):
+
+- mixed queue (tomo, command, tomo) runs in order; actions dispatched with correct kwargs
+- a job dict **without** a `kind` field still runs as a tomogram (back-compat)
+- multi-step command job runs its steps in order
+- idempotent resume re-runs from the top; non-idempotent resume prompts
+- `tomo_queue_show`, the GUI dialog, and the web generator all survive a command job
+  with no `params` key
+- a `move` to a device **not** on the allow-list is refused at add time *and* at
+  execution time
+
+### Suggested commit breakdown
+1. `feat(flomni): type-tag tomo queue jobs (kind field, back-compat default)`
+2. `feat(flomni): add command-job action registry, device allow-list and _run_command_job`
+3. `feat(flomni): tomo_queue_add_command + show/execute branching`
+4. `feat(flomni): publish action registry via tomo_queue_actions global var`
+5. `feat(bec_widgets): add-to-queue button and command-job builder in TomoQueueDialog`
+6. `feat(bec_widgets): status-keyed row locks and mid-run reordering in TomoQueueDialog`
+7. `fix(flomni): guard webpage generator against command jobs without params`
+
+---
+
+## 10. Open questions — resolved with Mirko, one detail still TBD
+
+**Resolved (this round):**
+
+1. **The final action set**: `move` + `optimize_idgap` only, deliberately minimal.
+   No `foptics_in`/`foptics_out`/`feye_in`/`feye_out`/shutter in Step 2, despite the
+   first three already existing as callable methods — explicitly deferred to future
+   "dedicated scripts" not yet written. No energy-change wrapper — energy changes go
+   through `move` on `mokev`.
+2. **The curated device allow-list** for `move`: `mokev` and `idgap`. Both live in
+   `bl_frontend.yaml`, not the flomni sim config — `move` on either is only
+   exercisable against the real front end.
+3. `optimize_idgap`'s `search_range` default/bounds (0.5mm default, 0–2mm) confirmed
+   as roughly right placeholders.
+4. `move` kwargs shape: nested `{"positions": {name: target}}`, as specced.
+
+**Still open — deferred by Mirko to the implementation session, not blocking on it
+now:**
+
+- What signal/device `optimize_idgap` should scan/read to locate "the peak".
+- `idgap`'s unit (`mm` assumed in §3.1, not explicitly reconfirmed).
+
+Resolved since the earlier draft: registry distribution (global var, §7);
+non-idempotent resume behaviour (prompt, §5); multi-step jobs (yes, §2); ordering
+constraints (jobs are independent and run in order; the executor's resume-before-fresh
+rule handles the one real hazard).
diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_TESTING.md b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_TESTING.md
new file mode 100644
index 0000000..d47b9b4
--- /dev/null
+++ b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/TOMO_QUEUE_TESTING.md
@@ -0,0 +1,352 @@
+# Tomo Queue — Executor Rework: Testing & Continuity
+
+**Read this first.** It is the handoff for the tomo-queue work: what changed, what
+must be tested against the simulated flOMNI, and what is deliberately *not* built
+yet. Companion to `TOMO_QUEUE_COMMAND_JOBS_PLAN.md` (the design for the *next*
+step — command jobs — which is **not** implemented).
+
+**Status:** Step 1 (executor rework) implemented in `flomni.py`. **Sections A, B, and
+C of the checklist have now passed against the live sim** (10/10 tests, run together
+with no cross-test interference — see §4). **All of §6's open questions are now
+resolved with Mirko** (registry/allow-list in `TOMO_QUEUE_COMMAND_JOBS_PLAN.md` §10;
+the `_TOMO_SCAN_PARAM_NAMES` rename is done) — **Step 2 is unblocked to start.** Only
+**D (real instrument)**, needing beam time, remains open. Steps 2 and 3 are designed
+but not written.
+
+---
+
+## 0. House rules (apply to every change here)
+
+- **Minimal diffs.** No incidental Black reformats of unrelated lines. `flomni.py`
+  has pre-existing formatting drift (5 `black` hunks that predate this work) —
+  **leave it alone**; do not "fix" it in the same commit.
+- Validate every edited `.py` with `ast.parse` and
+  `black --check --line-length=100 --skip-magic-trailing-comma`. The correct check
+  is *"no NEW black hunks vs. the unmodified file"*, not "black is clean".
+- Targeted `str_replace`-style edits, not file rewrites.
+- Conventional Commits.
+- **Sim/hardware validation is the final gate.** Unit tests and mock harnesses come
+  first, but nothing is "verified" until it has run against the simulated (and
+  ultimately real) instrument.
+
+---
+
+## 1. What changed (Step 1 — already in `flomni.py`)
+
+The tomo queue executor was reworked from a **fixed-index loop** to an
+**id-based pick-next loop**. Diff is +100/−17 across 5 hunks:
+
+1. `import uuid`.
+2. `_TomoQueueProxy` gained:
+   - `update_by_id(job_id, **kwargs) -> bool` — write status by stable id, not list
+     index. Returns `False` if the id is gone (job deleted concurrently).
+   - `ensure_ids() -> list` — assign a uuid to any job lacking one, persist. Heals
+     legacy queues written before the `id` field existed.
+3. `tomo_queue_add()` now stamps `"id": uuid.uuid4().hex` on each new job.
+4. `tomo_queue_execute()` rewritten: re-reads the queue from its global var at the
+   start of **every** job (no up-front snapshot) and addresses jobs by `id`.
+
+### Why (the motivation — don't undo this)
+
+The old loop bound `for i in range(start_index, len(jobs))` against a snapshot taken
+once, then wrote status back **by index** via `update(i, ...)`, which reloaded the
+*live* var. If the queue was edited while running (GUI, or a second CLI session),
+indices drifted: status writes landed on the **wrong job**, or `update()` raised
+`IndexError`. Appended jobs were also never picked up by the running loop.
+
+The pick-next loop makes it safe to **append, delete a pending job, and reorder the
+pending tail while the queue is running**. Those edits take effect at the next job
+boundary.
+
+### Two deliberate behaviours (these are not bugs)
+
+- **Resume-before-fresh.** If any job is `incomplete`/`running`, it runs **before**
+  any fresh `pending` job, regardless of list order. Tomo progress lives in ONE
+  shared `progress` global var — starting a fresh `tomo_scan()` ahead of an
+  interrupted one would overwrite the very state `tomo_scan_resume()` needs. The old
+  strictly-in-order loop made this impossible; allowing reorder reintroduces the
+  risk, hence the explicit guard.
+- **`start_index` semantics changed.** It is now "position in the *current* order
+  below which fresh `pending` jobs are ignored", applied at each pick. Resumable jobs
+  **ignore** `start_index` (a crashed job must be recovered wherever it sits).
+  Default `0` behaves as before.
+
+### Invariant to preserve
+
+**At most one job may ever be `running`/`incomplete` at a time**, because `progress`
+is a single global var. Nothing in append/delete/reorder can create a second one
+(only execution sets those statuses). Do not add a "mark incomplete manually"
+affordance without revisiting this.
+
+### What Step 1 does NOT do
+
+No `kind` field, no command jobs, no GUI changes, no webpage-generator changes.
+A queue of plain tomo jobs behaves exactly as before, plus the ids.
+
+---
+
+## 2. Offline validation already done
+
+`ast.parse` clean; **no new black hunks** vs. the unmodified file. A mock-client
+harness (extracting the *real* `_TomoQueueProxy` + `tomo_queue_execute` source via
+AST, not a reimplementation) passes 10 scenarios: in-order run; done-jobs skipped;
+id-less legacy jobs healed; resume-before-fresh; `running` treated as resumable;
+reorder / delete / append mid-run; failure → `incomplete` + re-raise + correct
+resume on re-run; `update_by_id` found/missing.
+
+**This proves the logic, not the integration.** The mock stubs `tomo_scan()`, so it
+cannot prove that params are really restored onto the live global vars the scan
+reads, nor that `tomo_scan_resume()` recovers real mid-scan `progress` state. That
+is what the sim is for.
+
+---
+
+## 3. THE TEST HARNESS — use `pytest_bec_e2e`, and mind the trap
+
+BEC already ships **`pytest_bec_e2e`** (registered `pytest11` entry point, in the
+`bec` repo). It provides a **live** `BECClient` — real server, real Redis, real
+global vars. Without `--start-servers` it **attaches to an already-running server**,
+which is exactly the sim session. Added to `csaxs_bec`'s `dev` extras
+(`pyproject.toml`) as `pytest-bec-e2e`.
+
+### ⚠️ TRAP — do not use `bec_client_lib` as-is
+
+Both shipped client fixtures (`bec_client_lib_with_demo_config`,
+`bec_ipython_client_with_demo_config`) call:
+
+```python
+bec.config.load_demo_config(force=True)
+```
+
+Against a running sim session this **overwrites the loaded simulated-flOMNI device
+config with BEC's demo devices**. Write our own fixture that reuses the plumbing
+(`bec_servers`, `bec_redis_fixture`, `bec_services_config_file_path`) but either
+loads `device_configs/simulated_omny/simulated_flomni.yaml` explicitly or assumes the
+running session already has it.
+
+### The fixture: `tests/e2e/_bootstrap.py` + `tests/e2e/conftest.py` → `flomni_sim` — BUILT
+
+This is the reusable piece — worth more than the checklist itself. It's built and
+factored into two files so section B/C tests (which need a real, separate OS process
+to kill or run concurrently — see below) can build the same `Flomni` outside pytest:
+
+- `_bootstrap.py`: `build_flomni(services_config_path) -> (bec, flomni)` — connects
+  (no demo-config load), does the builtins/reload bootstrap (see gotcha below),
+  neutralizes side effects, and does the one-time RT-feedback/fsamx setup (see
+  gotchas below). Also `neutralize_side_effects()` and `shutdown(bec)`.
+- `conftest.py`: the `flomni_sim` pytest fixture, which calls `build_flomni()` and
+  snapshots/restores the `tomo_queue` / `tomo_progress` global vars around each test.
+- `_run_queue_subprocess.py`: a standalone script (`python _run_queue_subprocess.py
+  <services_config_path> [start_index]`) that calls `build_flomni()` +
+  `tomo_queue_execute()`. Used by every section B/C test as the thing that gets
+  `SIGKILL`ed or run concurrently with a second client.
+- `_queue_helpers.py`: `SHORT_SCAN_PARAMS` / `FAST_STEPSIZE` / `add_short_job()` for
+  building fast test tomograms; `ProgressSampler` (see thread gotcha below);
+  `spawn_queue_subprocess()`; `wait_until()`.
+
+Every future piece of work (command jobs, the GUI queue dialog, LamNI parity) can
+reuse all four.
+
+### Gotchas found getting a real scan to run (not obvious from reading `flomni.py`)
+
+None of these are Step-1 bugs — they're pre-existing `tomo_scan()`/environment
+dependencies that just hadn't been exercised end-to-end on this dev machine before.
+All are handled once, in `_bootstrap.py`:
+
+- **`BECIPythonClient`'s live-update machinery requires the main thread.**
+  `scans.umv(...)` (and anything else that goes through
+  `ipython_live_updates.process_request`) installs a `SIGINT` handler per request,
+  which Python only allows from a process's real main thread. So `tomo_queue_execute()`
+  itself can **never** run on a background Python thread — for section A's
+  "observe progress while it runs" this means *sampling* runs on a background thread
+  while execution stays on the main thread (`ProgressSampler`); for section B/C's
+  "kill it" / "run two of these at once" this means each session is a real OS
+  subprocess, not a thread.
+- **The builtins/reload bootstrap is order-sensitive.** `flomni.py`'s module-level
+  `def umv(*args): return scans.umv(...)` resolves `scans` from the *module's own
+  globals*, populated only by the `if builtins.__dict__.get("bec") is not None:`
+  block at import time. If `flomni.py` gets imported before `builtins.bec` is set
+  (likely, transitively), that block never runs and `umv` `NameError`s at call time.
+  Fix: set `builtins.bec/dev/scans` first, then `importlib.reload()` the module.
+- **`Flomni.__init__` has real side effects that must be neutralized for tests:**
+  `FlomniWebpageGenerator.start()` opens a local port and POSTs to a real external
+  URL; `tomo_scan()` unconditionally sends a SciLog summary at the end of every scan
+  (not configured on this dev machine) — both patched to no-ops.
+- **RT feedback + `fsamx` positioning is a real one-time hardware-setup step,**
+  normally folded into x-ray eye alignment (which tests skip): `fsamx` must be near
+  its configured "in" position before `feedback_enable_with_reset()` will run, and
+  `feedback_enable_with_reset()` itself locks `fsamx` read-only as its own last step
+  (the RT controller's PID correction owns fine positioning after that) — so a second
+  call must check `feedback_is_running()` first, or a plain `umv(fsamx, ...)` raises
+  `DisabledDeviceError`. Also expensive (re-zeros the interferometers), so skip it
+  entirely once feedback is already running from a previous session.
+- **Local output directories aren't created automatically:** `tomo_scan()` writes to
+  `~/data/raw/{documentation,logs,webpage,analysis/online/ptycho}` unconditionally
+  (PDF report, scan-number log, ...) — these didn't exist on this dev machine and had
+  to be created once (`mkdir -p`), not part of the repo.
+
+### Making scans short enough to test
+
+`sim_point_dwell_s` defaults to 20 ms. `SHORT_SCAN_PARAMS` in `_queue_helpers.py`
+sets `single_point_instead_of_fermat_scan=True` (one acquisition per angle instead of
+a full Fermat grid — `tomo_acquire_at_angle()`, not `tomo_scan_projection()`), small
+`fovx`/`fovy`, large `tomo_shellstep`, and a small `tomo_countingtime`. Combined with
+a coarse `tomo_angle_stepsize` (`FAST_STEPSIZE = 170.0` → 1 projection/subtomo, 8
+total) a full tomogram runs in roughly a minute — long enough to still be
+interruptible mid-scan for the crash-resume tests (§4, B5).
+
+---
+
+## 4. THE CHECKLIST
+
+### A. Sim, no crash needed — ✅ PASSED (`tests/e2e/test_tomo_queue.py`)
+
+1. **✅ Params actually restored per job.** `test_params_restored_per_job`. Two jobs
+   with distinct `tomo_angle_stepsize` (89.9 → N=2, 59.9 → N=3); a background thread
+   polls the shared `tomo_progress` global var (execution itself must stay on the
+   main thread — see the threading gotcha in §3) while each job runs and confirms
+   `progress["subtomo_total_projections"]`, computed live by `sub_tomo_scan()` from
+   the *live* global var, matches each job's own value in the right order. This is
+   the "did the SCAN actually use them" observable, agreed before running it.
+2. **✅ Legacy queue migration.** `test_legacy_queue_migration`. Two id-less jobs
+   written via `set_global_var("tomo_queue", [...])` ran cleanly; afterward every job
+   carried a unique healed `id`.
+3. **✅ Empty / all-done queue.** `test_empty_and_all_done_queue_are_noops`. Both
+   paths print their no-op message, never call `OMNYTools.yesno`, never scan.
+4. **✅ `start_index`.** `test_start_index_semantics`. `start_index=1` skipped a
+   pending job 0; marking it `incomplete` and re-running with the same `start_index`
+   ran it anyway (resumable jobs ignore `start_index`).
+
+### B. Sim, crash / interrupt paths — ✅ PASSED
+
+5. **✅ Real crash-resume.** `test_real_crash_resume`
+   (`tests/e2e/test_tomo_queue_crash_subprocess.py`). `tomo_queue_execute()` run as a
+   real OS subprocess (required — see §3), `SIGKILL`ed mid-job1. Confirmed via the
+   subprocess's own log: killed at `subtomo 3, angle 45.0`; the resume subprocess
+   printed `Resuming tomo scan at subtomo 3, angle 45.000 deg` and re-acquired that
+   *exact* subtomo/angle (checked by polling `tomo_progress` for a new `heartbeat`
+   after the resume subprocess starts, then comparing the full snapshot) before the
+   rest of the queue completed. Assertion design agreed beforehand.
+6. **✅ Exception mid-scan → pause.** `test_exception_mid_scan_pauses_queue`
+   (`tests/e2e/test_tomo_queue_crash.py`). `tomo_scan` patched to raise on its first
+   call only. Job → `incomplete`, exception re-raised out of `tomo_queue_execute()`,
+   later job stayed `pending`; re-running resumed the failed job first.
+7. **✅ Resume-before-fresh, explicitly.** `test_resume_before_fresh_explicit`
+   (same file). An `incomplete` job below a fresh `pending` one in list order ran
+   first regardless of position (tracked via an `update_by_id` call-order spy).
+
+### C. Concurrency — the path the GUI will use — ✅ PASSED (`tests/e2e/test_tomo_queue_concurrency.py`)
+
+8. **✅ Edit the queue from a second client while one runs.**
+   `test_concurrent_queue_edits_from_second_client`. Session A = subprocess executing
+   4 jobs; session B = the pytest process's own client, mid-job1: reordered the
+   pending tail (job4 ahead of job2), deleted job3, appended job5 — all via the same
+   redis-backed `tomo_queue` global var, no live scan-param vars touched. Session A
+   picked up every change at the next job boundary: final run order was job1, job4,
+   job2, job5, with job3 never appearing.
+9. **✅ Delete the *running* job mid-run.** `test_delete_running_job_mid_run`.
+   Deleted job1 (index 0, `status="running"`) out from under the executing
+   subprocess. Degraded gracefully: no crash (`update_by_id` on the vanished id
+   just returns `False`), job1's scan finished on its own terms, job2 still ran and
+   was marked `done` correctly.
+
+### D. Real instrument (not sim) — NOT DONE, needs Mirko / beam time
+
+10. One queued tomogram end-to-end on the real instrument, to confirm the sim's
+    timing/progress semantics match reality closely enough that resume lands on the
+    right angle.
+
+**Gate:** A–C passed on the sim (10/10 tests, run together, no cross-test
+interference — ~18 min). **Step 1 is verified enough to build Steps 2/3 on.** D and
+§6 remain open before Step 2 work should actually start.
+
+---
+
+## 5. Steps 2 & 3 — designed, NOT implemented
+
+Full design for command jobs is in `TOMO_QUEUE_COMMAND_JOBS_PLAN.md`. Decisions
+settled since that document was written — **it is not fully up to date; this section
+supersedes it where they conflict**:
+
+### Command jobs (Step 2)
+
+- **Named-action registry.** No `exec` of stored code strings — rejected as
+  unauditable, fails at run time, breaks crash-resume.
+- **Multi-step jobs.** A command job holds a **list** of steps, not a single action,
+  so "optics out → change energy → optics in" is ONE queue row:
+  ```python
+  {"kind": "command", "label": "...", "steps": [{"action": ..., "kwargs": {...}}, ...],
+   "idempotent": bool, "status": ..., "id": ..., "added_at": ...}
+  ```
+- **No per-step resume cursor.** On resume: if all steps are idempotent, re-run the
+  whole sequence from the top (re-doing an absolute move is harmless). If any step is
+  non-idempotent, the job is flagged non-idempotent and resume **prompts** the
+  operator (`OMNYTools.yesno`) rather than silently re-running.
+- **Curated device allow-list** for the generic `move` action. NOT "any device in
+  `dev`". This is the difference between a reconfiguration tool and a remote control
+  for the whole endstation.
+- **Registry entry = 4 things:** bound method name, `idempotent` default, help string,
+  and a **params schema** (possibly empty).
+- **Result-dependent logic lives INSIDE a named action, never across queue steps.**
+  E.g. "scan idgap, move to peak" is ONE action (`optimize_idgap`) whose method does
+  the scan, finds the peak, and moves — *not* three queue steps. The queue therefore
+  never has to carry runtime values between steps; `kwargs` stays plain JSON. A
+  compound action may take parameters (e.g. `search_range`), declared in its params
+  schema like any other.
+- **Publish the registry via a global var** (`tomo_queue_actions`) on `Flomni.__init__`
+  so the GUI builds its dropdowns/forms from it — no hardcoded mirror to drift (the
+  `QUEUE_PARAM_NAMES` mirror between `flomni.py` and `tomo_params.py` already has this
+  problem).
+- **Back-compat:** a job with no `kind` is treated as `"tomo"`.
+
+### GUI (Step 3) — `tomo_params.py` / `TomoQueueDialog`
+
+Agreed interaction model:
+
+- **Two separate write paths.** "Done" writes the **live param global vars** → must be
+  **blocked while a scan is running** (the scan re-reads those vars per projection, so
+  editing them mid-run perturbs the running acquisition). "Add to queue" writes **only
+  the `tomo_queue` var** → always allowed, even mid-run. *This separation is the whole
+  reason the GUI is usable during a run — do not let "Add to queue" touch live vars.*
+- "Add to queue" **cannot** call the CLI `tomo_queue_add()` — that snapshots the
+  **live** vars and would ignore the operator's unapplied form edits. The dialog must
+  assemble the job dict itself (edited fields + current live values for params the form
+  doesn't expose) so the snapshot is complete.
+- **No in-place row editing.** Editing a queued job = delete it + add a new one +
+  reorder into place. Keeps one construction path for jobs.
+- **Row locks keyed on `status`, not on a global busy signal:**
+  - `running` row: not draggable, not deletable. Deletable only after stop (then it is
+    `incomplete`).
+  - `pending` / `incomplete`: freely draggable and deletable.
+  - `done`: pinned (harmless to move, but confusing to read).
+- **Reorder is allowed mid-run**, and takes effect at the next job boundary. Reject any
+  drop that would move a row **above** the running row — the running row is a hard floor
+  for drops, so the visual order stays honest about what runs next.
+- A **command-job builder pop-up** (structured form, not a text editor): action
+  dropdown from the registry, fields from the action's params schema, live validation
+  against real device limits, OK disabled until valid. **No syntax-checked code box** —
+  syntax-valid is nearly worthless as a safety check (`umv(dev.fsamx, -1e9)` parses
+  fine), and it gives false confidence.
+- Execution stays **CLI-only**; the GUI server process can't run the blocking scan loop.
+
+### Webpage generator (`flomni_webpage_generator.py`)
+
+Not yet touched. When command jobs land, any code doing `job["params"][...]` will
+`KeyError` on a command job (no `params` key) — guard with
+`job.get("kind", "tomo") == "tomo"`. Also re-check projection/ETA counting so command
+jobs don't skew tomogram-based estimates.
+
+---
+
+## 6. Open questions for Mirko — all resolved
+
+- **The final action set and device allow-list**, see `TOMO_QUEUE_COMMAND_JOBS_PLAN.md`
+  §10 — `move` + `optimize_idgap` only, `_TOMO_QUEUE_MOVE_DEVICES = {mokev, idgap}`.
+  One detail (`optimize_idgap`'s peak-detection signal) deferred to the
+  implementation session, not blocking.
+- **`_TOMO_QUEUE_PARAM_NAMES` renamed to `_TOMO_SCAN_PARAM_NAMES`** (it reads live
+  properties, not queue state — the old name was misleading). Done in `flomni.py`;
+  the `tomo_params.py` comment referencing it by name was updated too (the
+  `QUEUE_PARAM_NAMES` mirror constant itself keeps its own name, unchanged).
+
+Nothing left blocking Step 2 except D (real instrument, needs beam time).
diff --git a/tests/e2e/_bootstrap.py b/tests/e2e/_bootstrap.py
new file mode 100644
index 0000000..40161d8
--- /dev/null
+++ b/tests/e2e/_bootstrap.py
@@ -0,0 +1,113 @@
+"""Shared bootstrap for constructing a live Flomni against the running sim.
+
+Used both by conftest.py (in-process, section A) and by
+_run_queue_subprocess.py (spawned as a separate OS process, section B/C) --
+pulled out here so both paths for "build a real Flomni pointed at the sim"
+share one definition of what real-instrument setup and side-effect
+neutralization is needed. See TOMO_QUEUE_TESTING.md section 3.
+
+Deliberately does NOT call bec.config.load_demo_config()/
+update_session_with_file(): assumes the running sim session already has
+device_configs/simulated_omny/simulated_flomni.yaml loaded.
+"""
+
+from __future__ import annotations
+
+import builtins
+import importlib
+
+from bec_ipython_client import BECIPythonClient
+from bec_lib.messaging_services import SciLogMessagingService
+from bec_lib.redis_connector import RedisConnector
+from bec_lib.service_config import ServiceConfig
+
+from csaxs_bec.bec_ipython_client.plugins.flomni import flomni as flomni_module
+from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
+    FlomniWebpageGenerator,
+)
+from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import OMNYTools
+
+
+class _NoOpSciLogMessage:
+    """Chainable no-op standing in for a real SciLog message object."""
+
+    def add_text(self, *a, **k):
+        return self
+
+    def add_tags(self, *a, **k):
+        return self
+
+    def add_attachment(self, *a, **k):
+        return self
+
+    def send(self, *a, **k):
+        return self
+
+
+def neutralize_side_effects() -> None:
+    """Patch away operator-facing/networked side effects of Flomni().
+
+    Plain class-attribute assignment, not pytest's monkeypatch: every
+    caller (in-process fixture, subprocess script) wants the same patches
+    for the life of the process, so there's nothing to revert.
+
+    - OMNYTools.yesno -> always answers yes, so tomo_queue_execute()'s
+      confirmation prompt doesn't hang a headless run.
+    - FlomniWebpageGenerator.start -> no-op. Flomni.__init__ normally starts
+      this unconditionally, which opens a local HTTP server and spawns
+      background threads that POST to a real external URL.
+    - SciLogMessagingService.new -> no-op. tomo_scan() unconditionally sends
+      a SciLog summary at the end of every scan; SciLog isn't
+      configured/reachable in this dev environment.
+    """
+    OMNYTools.yesno = lambda self, *a, **k: True
+    FlomniWebpageGenerator.start = lambda self: None
+    SciLogMessagingService.new = lambda self, *a, **k: _NoOpSciLogMessage()
+
+
+def build_flomni(services_config_path):
+    """Connect to the already-running sim and return (bec, flomni)."""
+    config = ServiceConfig(services_config_path)
+    bec = BECIPythonClient(config, RedisConnector, forced=True, wait_for_server=True)
+    bec.start()
+
+    # flomni.py's module-level `umv` resolves `scans` from the module's own
+    # globals, populated only by the `if builtins.__dict__.get("bec") is not
+    # None:` block at import time. Set builtins first, then reload so it
+    # re-runs (needed even on a fresh process: importing flomni_module above
+    # ran that block once already, before bec existed).
+    builtins.__dict__["bec"] = bec
+    builtins.__dict__["dev"] = bec.device_manager.devices
+    builtins.__dict__["scans"] = bec.scans
+    importlib.reload(flomni_module)
+    builtins.__dict__["umv"] = flomni_module.umv
+
+    neutralize_side_effects()
+
+    flomni = flomni_module.Flomni(bec)
+
+    # RT interferometer feedback + fsamx-in-position setup, mirroring
+    # Flomni._align_setup(). Only intervene when actually needed: fsamx is
+    # normally left read_only (locked) by a *previous*
+    # feedback_enable_with_reset() call (its own last step, so the RT
+    # controller's PID correction owns fine positioning going forward), and
+    # a move request against a read_only device raises DisabledDeviceError.
+    # feedback_enable_with_reset() itself is expensive (re-zeros the
+    # interferometers) -- skip it entirely if feedback is already running,
+    # which it normally is after the first caller in a test session sets it
+    # up (device-server state persists across client (re)connections).
+    rtx = bec.device_manager.devices.rtx
+    if not rtx.controller.feedback_is_running():
+        fsamx = bec.device_manager.devices.fsamx
+        fsamx_in = fsamx.user_parameter.get("in")
+        if abs(fsamx.readback.get() - fsamx_in) > 0.3:
+            fsamx.read_only = False
+            flomni_module.umv(fsamx, fsamx_in)
+        flomni.feedback_enable_with_reset()
+
+    return bec, flomni
+
+
+def shutdown(bec) -> None:
+    bec.shutdown()
+    bec._client._reset_singleton()
diff --git a/tests/e2e/_queue_helpers.py b/tests/e2e/_queue_helpers.py
new file mode 100644
index 0000000..b0b4063
--- /dev/null
+++ b/tests/e2e/_queue_helpers.py
@@ -0,0 +1,120 @@
+"""Shared helpers for driving tomo_queue_execute() against the live sim."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+
+_SUBPROCESS_SCRIPT = Path(__file__).with_name("_run_queue_subprocess.py")
+
+SHORT_SCAN_PARAMS = dict(
+    # Single acquisition per angle instead of a full Fermat grid -- makes a
+    # queued "tomogram" fast enough to run in a test.
+    single_point_instead_of_fermat_scan=True,
+    tomo_countingtime=0.05,
+    frames_per_trigger=1,
+    fovx=2,
+    fovy=2,
+    tomo_shellstep=1,
+    tomo_angle_range=180,
+    zero_deg_reference_at_each_subtomo=False,
+)
+
+# N = int(tomo_angle_range / tomo_angle_stepsize) = 1 -> 8 projections total,
+# the fastest a real tomogram can be for tests that don't care about the
+# per-job N itself, just that a job runs cleanly.
+FAST_STEPSIZE = 170.0
+
+
+def short_params(flomni, tomo_angle_stepsize: float = FAST_STEPSIZE) -> dict:
+    for name, value in SHORT_SCAN_PARAMS.items():
+        setattr(flomni, name, value)
+    flomni.tomo_angle_stepsize = tomo_angle_stepsize
+    return {name: getattr(flomni, name) for name in flomni._TOMO_SCAN_PARAM_NAMES}
+
+
+def add_short_job(flomni, label: str, tomo_angle_stepsize: float = FAST_STEPSIZE) -> int:
+    short_params(flomni, tomo_angle_stepsize)
+    return flomni.tomo_queue_add(label)
+
+
+class ProgressSampler:
+    """Samples ``flomni.progress[key]`` on a background thread while the
+    caller runs something blocking (e.g. ``tomo_queue_execute()``) in the
+    foreground.
+
+    ``tomo_queue_execute()`` itself must stay on the main thread:
+    BECIPythonClient's live-update machinery installs a SIGINT handler per
+    scan request (``ipython_live_updates.process_request``), and Python only
+    allows ``signal.signal()`` from the main thread of the main interpreter
+    -- calling it from a worker thread raises ``ValueError``. So it's the
+    *sampling*, not the execution, that runs in the background here.
+    """
+
+    def __init__(self, flomni, key: str = "subtomo_total_projections", interval: float = 0.05):
+        self._flomni = flomni
+        self._key = key
+        self._interval = interval
+        self._observations: list = []
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._run, daemon=True)
+
+    def _run(self):
+        while not self._stop.is_set():
+            self._observations.append(self._flomni.progress.get(self._key))
+            time.sleep(self._interval)
+
+    def __enter__(self) -> "ProgressSampler":
+        self._thread.start()
+        return self
+
+    def __exit__(self, *exc_info) -> None:
+        self._stop.set()
+        self._thread.join(timeout=5)
+
+    @property
+    def observations(self) -> list:
+        return list(self._observations)
+
+    def first_index_of(self, value) -> int | None:
+        try:
+            return self._observations.index(value)
+        except ValueError:
+            return None
+
+
+def spawn_queue_subprocess(
+    services_config_path, log_path, start_index: int = 0
+) -> subprocess.Popen:
+    """Launch _run_queue_subprocess.py as a real OS process against the same
+    sim, so it can be SIGKILLed -- or run concurrently with a second client
+    editing the queue -- the way a real crashed/second kernel would be.
+
+    Output goes to ``log_path``, not a PIPE: a killed/long-running subprocess
+    whose stdout pipe nobody drains can deadlock once the OS pipe buffer
+    fills (tomo_queue_execute() prints a lot -- progress bars per move).
+    """
+    with open(log_path, "w") as log_file:
+        # Popen dup()s the fd for the child; safe to close our copy right
+        # after spawning instead of leaking it for the subprocess's lifetime.
+        return subprocess.Popen(
+            [sys.executable, str(_SUBPROCESS_SCRIPT), str(services_config_path), str(start_index)],
+            stdout=log_file,
+            stderr=subprocess.STDOUT,
+        )
+
+
+def wait_until(predicate, timeout: float, interval: float = 0.1):
+    """Poll ``predicate()`` until it returns a truthy value or ``timeout``
+    elapses. Returns the truthy value, or raises TimeoutError.
+    """
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = predicate()
+        if result:
+            return result
+        time.sleep(interval)
+    raise TimeoutError(f"condition not met within {timeout}s")
diff --git a/tests/e2e/_run_queue_subprocess.py b/tests/e2e/_run_queue_subprocess.py
new file mode 100644
index 0000000..06d93b4
--- /dev/null
+++ b/tests/e2e/_run_queue_subprocess.py
@@ -0,0 +1,36 @@
+"""Standalone script: connect to the running sim and call
+flomni.tomo_queue_execute(). Run as a real OS subprocess (its own main
+thread) rather than a Python thread inside the test process: BECIPythonClient
+'s live-update machinery installs a SIGINT handler per scan request, which
+Python only allows from a process's real main thread (see
+_queue_helpers.ProgressSampler for the same constraint hit from the thread
+side). Section B/C tests spawn this and SIGKILL it (or edit the queue
+concurrently from a second client) to simulate a crashed kernel / a second
+operator session.
+
+Usage: python _run_queue_subprocess.py <services_config_path> [start_index]
+Exit code is 0 on a clean run, 1 if tomo_queue_execute() raised.
+"""
+
+import sys
+
+from _bootstrap import build_flomni, shutdown
+
+
+def main() -> int:
+    services_config_path = sys.argv[1]
+    start_index = int(sys.argv[2]) if len(sys.argv) > 2 else 0
+
+    bec, flomni = build_flomni(services_config_path)
+    try:
+        flomni.tomo_queue_execute(start_index=start_index)
+    except Exception as exc:  # noqa: BLE001 - reported via exit code, not re-raised
+        print(f"_run_queue_subprocess: tomo_queue_execute() raised: {exc}", file=sys.stderr)
+        return 1
+    finally:
+        shutdown(bec)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 0000000..ca1f236
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -0,0 +1,37 @@
+"""Fixtures for end-to-end tomo-queue tests against a running flOMNI sim.
+
+Deliberately does NOT use the shipped ``bec_client_lib_with_demo_config`` /
+``bec_ipython_client_with_demo_config`` fixtures from pytest_bec_e2e: both
+call ``bec.config.load_demo_config(force=True)``, which against a running sim
+session overwrites the loaded simulated-flOMNI device config with BEC's demo
+devices. See TOMO_QUEUE_TESTING.md section 3.
+"""
+
+import pytest
+from _bootstrap import build_flomni, shutdown
+
+_QUEUE_VAR = "tomo_queue"
+_PROGRESS_VAR = "tomo_progress"
+
+
+@pytest.fixture
+def flomni_sim(bec_redis_fixture, bec_services_config_file_path, bec_servers):
+    """A live ``Flomni`` instance attached to the already-running sim session.
+
+    See ``_bootstrap.build_flomni()`` for what connecting/constructing this
+    involves and why. The ``tomo_queue`` / ``tomo_progress`` global vars are
+    snapshotted before the test and restored after, so tests don't leak
+    queue/progress state into each other or into a real operator's session
+    on the same sim.
+    """
+    bec, flomni = build_flomni(bec_services_config_file_path)
+
+    queue_snapshot = bec.get_global_var(_QUEUE_VAR)
+    progress_snapshot = bec.get_global_var(_PROGRESS_VAR)
+
+    try:
+        yield flomni
+    finally:
+        bec.set_global_var(_QUEUE_VAR, queue_snapshot)
+        bec.set_global_var(_PROGRESS_VAR, progress_snapshot)
+        shutdown(bec)
diff --git a/tests/e2e/test_smoke.py b/tests/e2e/test_smoke.py
new file mode 100644
index 0000000..b792f25
--- /dev/null
+++ b/tests/e2e/test_smoke.py
@@ -0,0 +1,8 @@
+"""Smoke test for the flomni_sim fixture -- construct it, touch no scan."""
+
+
+def test_flomni_sim_constructs_and_reads_params(flomni_sim):
+    flomni = flomni_sim
+    print("sample_name:", flomni.sample_name)
+    print("tomo_countingtime:", flomni.tomo_countingtime)
+    flomni.tomo_queue_show()
diff --git a/tests/e2e/test_tomo_queue.py b/tests/e2e/test_tomo_queue.py
new file mode 100644
index 0000000..a1a3f96
--- /dev/null
+++ b/tests/e2e/test_tomo_queue.py
@@ -0,0 +1,159 @@
+"""Checklist section A -- sim, no crash needed. See TOMO_QUEUE_TESTING.md."""
+
+import datetime
+
+from _queue_helpers import ProgressSampler
+from _queue_helpers import add_short_job as _add_short_job
+from _queue_helpers import short_params as _short_params
+
+
+def test_params_restored_per_job(flomni_sim):
+    """A1: each job's snapshotted params, not just whatever the global var
+    last held, actually reach the running scan.
+
+    job1 and job2 differ only in tomo_angle_stepsize. sub_tomo_scan() derives
+    progress["subtomo_total_projections"] = int(tomo_angle_range /
+    tomo_angle_stepsize) live, at scan time, from the *live* global var --
+    so if the queue's per-job setattr() restore didn't really work, both
+    jobs would show whatever stepsize happened to be live last (i.e. job2's,
+    since it was set most recently), not each job's own value.
+    """
+    flomni = flomni_sim
+
+    job1_stepsize = 89.9
+    job2_stepsize = 59.9
+    expected_n = {"job1": int(180 / job1_stepsize), "job2": int(180 / job2_stepsize)}
+    assert expected_n["job1"] != expected_n["job2"], "test setup must pick distinguishable N"
+
+    _add_short_job(flomni, "job1", job1_stepsize)
+    _add_short_job(flomni, "job2", job2_stepsize)
+
+    # Reset first: leftover progress state from a previous run could
+    # otherwise coincidentally satisfy the "value observed" check below
+    # before job1 has actually started.
+    flomni.progress.reset()
+
+    with ProgressSampler(flomni) as sampler:
+        flomni.tomo_queue_execute()
+
+    observed = sampler.observations
+    job1_idx = sampler.first_index_of(expected_n["job1"])
+    job2_idx = sampler.first_index_of(expected_n["job2"])
+
+    assert job1_idx is not None, (
+        f"job1 (stepsize={job1_stepsize}) should have shown "
+        f"subtomo_total_projections={expected_n['job1']} at some point; observed {observed}"
+    )
+    assert job2_idx is not None, (
+        f"job2 (stepsize={job2_stepsize}) should have shown "
+        f"subtomo_total_projections={expected_n['job2']} at some point; observed {observed}"
+    )
+    assert job1_idx < job2_idx, (
+        "job1's N should be observed before job2's -- jobs ran out of order "
+        f"(job1 first at {job1_idx}, job2 first at {job2_idx})"
+    )
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    assert [j["status"] for j in jobs] == ["done", "done"]
+
+
+def test_empty_and_all_done_queue_are_noops(flomni_sim, monkeypatch, capsys):
+    """A3: tomo_queue_execute() on an empty queue, and on a queue whose jobs
+    are all "done", must print a no-op message, never prompt, and never
+    trigger a scan.
+    """
+    flomni = flomni_sim
+
+    prompt_calls = []
+    from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import OMNYTools
+
+    monkeypatch.setattr(
+        OMNYTools, "yesno", lambda self, *a, **k: prompt_calls.append((a, k)) or True
+    )
+
+    # -- empty queue --
+    flomni._tomo_queue_proxy.clear()
+    capsys.readouterr()  # discard anything buffered so far
+    flomni.tomo_queue_execute()
+    out = capsys.readouterr().out
+    assert "Tomo queue is empty." in out
+    assert prompt_calls == []
+
+    # -- all-done queue --
+    _add_short_job(flomni, "already_done")
+    job_id = flomni._tomo_queue_proxy.as_list()[0]["id"]
+    flomni._tomo_queue_proxy.update_by_id(job_id, status="done")
+    capsys.readouterr()
+    flomni.tomo_queue_execute()
+    out = capsys.readouterr().out
+    assert "No pending tomo queue jobs to run." in out
+    assert prompt_calls == []
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    assert len(jobs) == 1 and jobs[0]["status"] == "done"  # untouched by the no-op call
+
+
+def test_start_index_semantics(flomni_sim):
+    """A4: start_index is "position below which fresh pending jobs are
+    ignored" -- it does NOT apply to resumable (incomplete/running) jobs,
+    which always run regardless of position.
+    """
+    flomni = flomni_sim
+
+    _add_short_job(flomni, "job0")
+    _add_short_job(flomni, "job1")
+    jobs = flomni._tomo_queue_proxy.as_list()
+    job0_id, job1_id = jobs[0]["id"], jobs[1]["id"]
+
+    flomni.tomo_queue_execute(start_index=1)
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[job0_id]["status"] == "pending", "job0 should have been skipped by start_index"
+    assert jobs[job1_id]["status"] == "done", "job1 should have run"
+
+    # Mark job0 "incomplete" (as if it had crashed) and drop progress so its
+    # resume is a clean no-op ("No tomo scan in progress to resume"). This
+    # test is about *selection* (does start_index wrongly skip a resumable
+    # job?), not resume accuracy -- that's section B.
+    flomni._tomo_queue_proxy.update_by_id(job0_id, status="incomplete")
+    flomni.progress.reset()
+
+    flomni.tomo_queue_execute(start_index=1)
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert (
+        jobs[job0_id]["status"] == "done"
+    ), "job0 is resumable (incomplete) and must run despite start_index=1"
+    assert jobs[job1_id]["status"] == "done", "job1 (already done) should not have been re-run"
+
+
+def test_legacy_queue_migration(flomni_sim):
+    """A2: a queue written before jobs carried an ``id`` field (e.g. by an
+    older client version) must run cleanly, and ensure_ids() must heal every
+    job with a unique id as a side effect of tomo_queue_execute().
+    """
+    flomni = flomni_sim
+    bec = flomni.client
+
+    legacy_jobs = []
+    for label in ("legacy1", "legacy2"):
+        params = _short_params(flomni)
+        legacy_jobs.append(
+            {
+                # deliberately no "id" key -- pre-Step-1 job shape
+                "label": label,
+                "params": params,
+                "status": "pending",
+                "added_at": datetime.datetime.now().isoformat(),
+            }
+        )
+    bec.set_global_var("tomo_queue", legacy_jobs)
+
+    flomni.tomo_queue_show()
+    flomni.tomo_queue_execute()
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    assert [j["status"] for j in jobs] == ["done", "done"]
+    ids = [j.get("id") for j in jobs]
+    assert all(ids), f"every job should have been healed with an id, got {ids}"
+    assert len(set(ids)) == len(ids), f"healed ids should be unique, got {ids}"
diff --git a/tests/e2e/test_tomo_queue_concurrency.py b/tests/e2e/test_tomo_queue_concurrency.py
new file mode 100644
index 0000000..a822892
--- /dev/null
+++ b/tests/e2e/test_tomo_queue_concurrency.py
@@ -0,0 +1,120 @@
+"""Checklist section C -- concurrency, the path the GUI will use. See TOMO_QUEUE_TESTING.md.
+
+Session A (executing the queue) is always a subprocess -- tomo_queue_execute()
+must run on a process's real main thread (see _queue_helpers.ProgressSampler).
+Session B (editing concurrently) is the pytest process's own flomni_sim
+connection: a second, independent client hitting the same redis-backed
+tomo_queue global var, exactly like a second operator's kernel or the future
+GUI dialog would.
+"""
+
+import datetime
+import uuid
+
+from _queue_helpers import add_short_job, spawn_queue_subprocess, wait_until
+
+
+def test_concurrent_queue_edits_from_second_client(
+    flomni_sim, bec_services_config_file_path, tmp_path
+):
+    """C8: while session A executes the queue, session B reorders the
+    pending tail, deletes a pending job, and appends a new one, all mid-job1.
+    Session A must pick up every change at the next job boundary and write
+    statuses to the right jobs -- the real-redis version of what the offline
+    mock harness proved in memory (section 2 of the doc).
+    """
+    flomni = flomni_sim
+    bec = flomni.client
+
+    add_short_job(flomni, "job1")
+    add_short_job(flomni, "job2")
+    add_short_job(flomni, "job3")
+    add_short_job(flomni, "job4")
+    jobs = flomni._tomo_queue_proxy.as_list()
+    ids = {j["label"]: j["id"] for j in jobs}
+
+    proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "sessionA.log")
+    try:
+        wait_until(lambda: flomni._tomo_queue_proxy.as_list()[0]["status"] == "running", timeout=30)
+
+        # -- session B's edits, all mid-job1: reorder the pending tail
+        # (job4 ahead of job2), delete job3, append job5. Deliberately does
+        # NOT touch any live tomo_* param global var here (only the
+        # tomo_queue list itself) -- job1's own live scan in session A is
+        # reading those same params every projection; touching them mid-run
+        # would perturb the *running* job, which is a different hazard than
+        # the one this test is about (see TOMO_QUEUE_COMMAND_JOBS_PLAN's
+        # "Done must be blocked while running" rule for the GUI).
+        current = {j["id"]: j for j in bec.get_global_var("tomo_queue")}
+        reordered = [
+            current[ids["job1"]],  # the running job, left in place
+            current[ids["job4"]],
+            current[ids["job2"]],
+        ]
+        bec.set_global_var("tomo_queue", reordered)
+
+        job5 = dict(current[ids["job1"]])  # clone an existing job's params snapshot
+        job5.update(
+            id=uuid.uuid4().hex,
+            label="job5",
+            status="pending",
+            added_at=datetime.datetime.now().isoformat(),
+        )
+        flomni._tomo_queue_proxy.append(job5)
+
+        returncode = proc.wait(timeout=420)  # 4 real jobs x ~8 projections each
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.wait(timeout=10)
+
+    log_text = (tmp_path / "sessionA.log").read_text()
+    assert returncode == 0, f"session A failed (see log):\n{log_text[-4000:]}"
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    labels_in_order = [j["label"] for j in jobs]
+    statuses = {j["label"]: j["status"] for j in jobs}
+
+    assert "job3" not in labels_in_order, "job3 was deleted mid-run and must not reappear"
+    assert labels_in_order == [
+        "job1",
+        "job4",
+        "job2",
+        "job5",
+    ], f"queue should reflect the reorder+append made mid-run, got {labels_in_order}"
+    assert statuses == {"job1": "done", "job4": "done", "job2": "done", "job5": "done"}
+
+
+def test_delete_running_job_mid_run(flomni_sim, bec_services_config_file_path, tmp_path):
+    """C9: deleting the *running* job mid-run (unprotected at CLI level --
+    the GUI will block it, per section 5 of the doc) must degrade
+    gracefully: update_by_id() returns False, job1's scan finishes on its
+    own terms, nothing crashes, and no status gets written to the wrong job.
+    """
+    flomni = flomni_sim
+
+    add_short_job(flomni, "job1")
+    add_short_job(flomni, "job2")
+
+    proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "sessionA.log")
+    try:
+        wait_until(lambda: flomni._tomo_queue_proxy.as_list()[0]["status"] == "running", timeout=30)
+
+        # Delete the running job (index 0) out from under session A.
+        flomni.tomo_queue_delete(0)
+
+        returncode = proc.wait(timeout=180)
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.wait(timeout=10)
+
+    log_text = (tmp_path / "sessionA.log").read_text()
+    assert (
+        returncode == 0
+    ), f"session A should degrade gracefully, not crash (see log):\n{log_text[-4000:]}"
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    labels = [j["label"] for j in jobs]
+    assert labels == ["job2"], f"deleted job1 must not reappear, got {labels}"
+    assert jobs[0]["status"] == "done", "job2 should still have run after job1's scan finished"
diff --git a/tests/e2e/test_tomo_queue_crash.py b/tests/e2e/test_tomo_queue_crash.py
new file mode 100644
index 0000000..0df4161
--- /dev/null
+++ b/tests/e2e/test_tomo_queue_crash.py
@@ -0,0 +1,89 @@
+"""Checklist section B -- sim, crash/interrupt paths. See TOMO_QUEUE_TESTING.md.
+
+B5 (real SIGKILL crash-resume) lives in test_tomo_queue_crash_subprocess.py --
+it needs a real OS process to kill, unlike B6/B7 here.
+"""
+
+import pytest
+from _queue_helpers import add_short_job
+
+
+def test_exception_mid_scan_pauses_queue(flomni_sim, monkeypatch):
+    """B6: tomo_scan() raising mid-job must leave that job "incomplete",
+    re-raise out of tomo_queue_execute(), and leave later jobs untouched
+    ("pending"). Re-running afterward must resume the failed job first,
+    before the still-pending one.
+    """
+    flomni = flomni_sim
+
+    add_short_job(flomni, "job1")
+    add_short_job(flomni, "job2")
+    jobs = flomni._tomo_queue_proxy.as_list()
+    job1_id, job2_id = jobs[0]["id"], jobs[1]["id"]
+
+    call_count = {"n": 0}
+    original_tomo_scan = type(flomni).tomo_scan
+
+    def _raise_once(self, *a, **k):
+        call_count["n"] += 1
+        if call_count["n"] == 1:
+            raise RuntimeError("simulated mid-scan failure")
+        return original_tomo_scan(self, *a, **k)
+
+    monkeypatch.setattr(type(flomni), "tomo_scan", _raise_once)
+
+    with pytest.raises(RuntimeError, match="simulated mid-scan failure"):
+        flomni.tomo_queue_execute()
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[job1_id]["status"] == "incomplete", "failed job must be marked incomplete"
+    assert jobs[job2_id]["status"] == "pending", "later job must be untouched by the failure"
+
+    # tomo_scan() raised before setting progress["tomo_start_time"], so the
+    # resume attempt below is itself a clean no-op ("No tomo scan in
+    # progress to resume") -- this test is about *selection order*
+    # (resume-before-fresh), not resume accuracy (that's B5).
+    flomni.tomo_queue_execute()
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[job1_id]["status"] == "done", "the previously-failed job must have been retried"
+    assert jobs[job2_id]["status"] == "done", "the pending job must have run after it"
+    assert call_count["n"] == 2, "tomo_scan should have been called exactly once per attempt"
+
+
+def test_resume_before_fresh_explicit(flomni_sim, monkeypatch):
+    """B7: an "incomplete" job below a fresh "pending" job in list order
+    must still run first, regardless of position (new pick-next behavior;
+    the old strictly-in-order loop couldn't reorder around this at all).
+    """
+    flomni = flomni_sim
+
+    add_short_job(flomni, "fresh_pending")
+    add_short_job(flomni, "crashed_earlier")
+    jobs = flomni._tomo_queue_proxy.as_list()
+    fresh_id, crashed_id = jobs[0]["id"], jobs[1]["id"]
+
+    # crashed_earlier sits BELOW fresh_pending in list order, but is
+    # resumable -- it must be picked first regardless.
+    flomni._tomo_queue_proxy.update_by_id(crashed_id, status="incomplete")
+    flomni.progress.reset()  # so its "resume" is a clean, fast no-op
+
+    run_order = []
+    original_update_by_id = type(flomni._tomo_queue_proxy).update_by_id
+
+    def _track(self, job_id, **kwargs):
+        if kwargs.get("status") == "running":
+            run_order.append(job_id)
+        return original_update_by_id(self, job_id, **kwargs)
+
+    monkeypatch.setattr(type(flomni._tomo_queue_proxy), "update_by_id", _track)
+    flomni.tomo_queue_execute()
+
+    assert run_order == [
+        crashed_id,
+        fresh_id,
+    ], f"crashed_earlier (resumable) should run before fresh_pending, got order {run_order}"
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[fresh_id]["status"] == "done"
+    assert jobs[crashed_id]["status"] == "done"
diff --git a/tests/e2e/test_tomo_queue_crash_subprocess.py b/tests/e2e/test_tomo_queue_crash_subprocess.py
new file mode 100644
index 0000000..2f49ba3
--- /dev/null
+++ b/tests/e2e/test_tomo_queue_crash_subprocess.py
@@ -0,0 +1,89 @@
+"""Checklist section B5 -- real crash-resume via SIGKILL. See TOMO_QUEUE_TESTING.md.
+
+Needs a real OS process to kill: a SIGKILLed process never reaches
+tomo_queue_execute()'s except-block, so the running job's status is left
+stuck at "running" -- exactly the "kernel crashed mid-scan" scenario the doc
+calls for, and not reproducible by raising an exception in-process (that's
+B6, in test_tomo_queue_crash.py).
+"""
+
+import signal
+
+from _queue_helpers import add_short_job, spawn_queue_subprocess, wait_until
+
+
+def test_real_crash_resume(flomni_sim, bec_services_config_file_path, tmp_path):
+    flomni = flomni_sim
+
+    add_short_job(flomni, "job1")
+    add_short_job(flomni, "job2")
+    jobs = flomni._tomo_queue_proxy.as_list()
+    job1_id, job2_id = jobs[0]["id"], jobs[1]["id"]
+
+    flomni.progress.reset()
+
+    proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "run1.log")
+    try:
+        # Wait until job1 is genuinely mid-scan (subtomo 3 of 8, at
+        # FAST_STEPSIZE's N=1-projection-per-subtomo) before killing -- not
+        # the very first projection, so this is unambiguously a *resume*,
+        # not indistinguishable from a fresh start.
+        wait_until(lambda: (flomni.progress.get("subtomo") or 0) >= 3, timeout=90)
+
+        # One atomic snapshot -- reading subtomo/angle/heartbeat as three
+        # separate round-trips could catch values from different instants.
+        before = flomni.progress.as_dict()
+
+        proc.send_signal(signal.SIGKILL)
+        proc.wait(timeout=10)
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.wait(timeout=10)
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[job1_id]["status"] == "running", (
+        "a SIGKILLed job never reaches tomo_queue_execute()'s except-block, so "
+        f"it should be stuck at 'running', not {jobs[job1_id]['status']!r}"
+    )
+    assert jobs[job2_id]["status"] == "pending", "job2 must be untouched by job1's crash"
+
+    # Resume: a second, fresh subprocess (a stand-in for "operator restarts
+    # their kernel") should pick job1 up mid-scan, not restart it.
+    proc2 = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "run2.log")
+    try:
+        after = wait_until(
+            lambda: _snapshot_if_new_heartbeat(flomni, before["heartbeat"]), timeout=60
+        )
+        returncode = proc2.wait(timeout=180)
+    finally:
+        if proc2.poll() is None:
+            proc2.kill()
+            proc2.wait(timeout=10)
+
+    log_text = (tmp_path / "run2.log").read_text()
+    assert returncode == 0, f"resume subprocess failed (see log):\n{log_text[-4000:]}"
+
+    assert after["subtomo"] == before["subtomo"], (
+        f"resume should re-acquire the exact in-flight subtomo {before['subtomo']}, "
+        f"got {after['subtomo']}"
+    )
+    assert after["angle"] == before["angle"], (
+        f"resume should re-acquire the exact in-flight angle {before['angle']}, "
+        f"got {after['angle']}"
+    )
+
+    jobs = {j["id"]: j for j in flomni._tomo_queue_proxy.as_list()}
+    assert jobs[job1_id]["status"] == "done", "job1 should have completed after resuming"
+    assert jobs[job2_id]["status"] == "done", "job2 should have run after job1 finished"
+
+
+def _snapshot_if_new_heartbeat(flomni, heartbeat_before):
+    """Return a full progress snapshot the first time ``heartbeat`` differs
+    from its pre-kill value (i.e. a fresh `_tomo_scan_at_angle` attempt has
+    started), else None so wait_until() keeps polling.
+    """
+    snapshot = flomni.progress.as_dict()
+    if snapshot.get("heartbeat") != heartbeat_before:
+        return snapshot
+    return None