wip scheduler

debye_bec/bec_widgets/widgets/client.py

@@ -69,7 +69,7 @@ class DigitalTwin(RPCBase):
 
 
 class RestartServer(RPCBase):
-    """Main widget of server restart widget"""
+    """Main widget of server restart widget."""
 
     _IMPORT_MODULE = "debye_bec.bec_widgets.widgets.restart_server.restart_server"
 
@@ -115,3 +115,102 @@ class ScanControlXAS(RPCBase):
         """
         Take a screenshot of the dock area and save it to a file.
         """
+
+
+class Scheduler(RPCBase):
+    """Schedule, persist and execute a sequence of BEC scan/device commands."""
+
+    _IMPORT_MODULE = "debye_bec.bec_widgets.widgets.scheduler.scheduler"
+
+    @rpc_call
+    def run_schedule(self):
+        """
+        Run the schedule, continuing from wherever it last left off.
+        """
+
+    @rpc_call
+    def abort_schedule(self):
+        """
+        Request that the schedule stop after the current item.
+        """
+
+    @rpc_call
+    def reset_schedule(self):
+        """
+        Clear all execution state and start the schedule over from item 1.
+        """
+
+    @rpc_call
+    def get_status(self) -> "dict":
+        """
+        RPC-exposed: current schedule state, e.g. for another widget or a script.
+        """
+
+    @rpc_call
+    def add_item(
+        self,
+        command: "str",
+        index: "int | None" = None,
+        kind: "str" = "custom",
+        form_state: "dict | None" = None,
+    ) -> "str":
+        """
+        RPC-exposed: insert a new, PENDING command into the schedule.
+        Safe to call while the schedule is running.
+
+        Args:
+            command: command text, evaluated the same way as the existing
+                items (against `scans`/`dev`) once the schedule runs.
+            index: position to insert at (0 = first). Clamped so the item
+                can never land before something already running or
+                finished. Defaults to appending at the end.
+            kind/form_state: optional structured description of how
+                `command` was built (see `schedule_item.ScheduleItem`),
+                used to reopen the Edit dialog pre-filled. Leave as
+                defaults for a plain, hand-typed command.
+
+        Returns:
+            The new item's item_id.
+        """
+
+    @rpc_call
+    def edit_item(
+        self,
+        item_id: "str",
+        command: "str",
+        kind: "str" = "custom",
+        form_state: "dict | None" = None,
+    ):
+        """
+        RPC-exposed: change the command of an item that has not started
+        yet. Raises `RuntimeError` for an item that is already
+        running/finished.
+        """
+
+    @rpc_call
+    def delete_item(self, item_id: "str"):
+        """
+        RPC-exposed: remove an item that has not started yet. Raises
+        `RuntimeError` for an item that is already running/finished.
+        """
+
+    @rpc_call
+    def move_item(self, item_id: "str", new_index: "int"):
+        """
+        RPC-exposed: move an item that has not started yet to a new
+        position (0 = first, but never before something already
+        running/finished). Raises `RuntimeError` for an item that is
+        already running/finished.
+        """
+
+    @rpc_call
+    def move_item_up(self, item_id: "str"):
+        """
+        RPC-exposed: swap an item with the one directly before it.
+        """
+
+    @rpc_call
+    def move_item_down(self, item_id: "str"):
+        """
+        RPC-exposed: swap an item with the one directly after it.
+        """

debye_bec/bec_widgets/widgets/scheduler/endpoints.py

@@ -0,0 +1,61 @@
+"""
+Redis endpoint(s) for the schedule widget plugin.
+
+Design notes (why this looks the way it does)
+-----------------------------------------------
+BEC never touches Redis with raw keys/commands. Every channel is described
+by an `EndpointInfo` (endpoint string + message type + allowed operations,
+see `bec_lib.endpoints`), and `RedisConnector` enforces both: calling an
+operation that isn't in the endpoint's `MessageOp` raises
+`IncompatibleRedisOperation`, and passing a message that isn't an instance
+of the endpoint's declared `message_type` raises
+`IncompatibleMessageForEndpoint`. Plain string topics still work but are
+deprecated. So a "conforming" custom endpoint means building a real
+`EndpointInfo`, exactly like `bec_lib.endpoints.MessageEndpoints` does
+internally.
+
+Namespace: we use `EndpointType.USER` ("user/...”), the same prefix BEC's
+own `MessageEndpoints.scan_queue_schedule()` uses for user-writable,
+persisted data (as opposed to `internal/`, `public/`, etc.).
+
+Message type: we deliberately do NOT declare our own `BECMessage`
+subclass. BEC's msgpack codec (`bec_lib.codecs.BECMessageEncoder.decode`)
+resolves an incoming message's class by name via
+`getattr(bec_lib.messages, type_name)` - i.e. only classes physically
+defined inside `bec_lib.messages` are resolvable this way. A message class
+declared in plugin code would fail to deserialize unless you monkey-patch
+it into that module, which is exactly the kind of side-door this plugin
+is meant to avoid. Instead we reuse `bec_lib.messages.VariableMessage`,
+a first-class, exported message type built for carrying an arbitrary
+(msgpack-serializable) payload under `.value`.
+`bec_lib.script_executor.upload_script` persists a script's text in Redis
+the same way, for the same reason - it's the established BEC pattern for
+"a plugin needs to stash its own structured data in Redis".
+"""
+
+from __future__ import annotations
+
+from bec_lib.endpoints import EndpointInfo, EndpointType, MessageOp
+from bec_lib.messages import VariableMessage
+
+
+def schedule(schedule_name: str) -> EndpointInfo:
+    """
+    Endpoint for one named widget schedule (an ordered list of commands
+    plus their execution status). The whole schedule is stored as a single
+    `VariableMessage` document that gets overwritten and republished on
+    every change (`MessageOp.SET_PUBLISH`), so any other subscriber -
+    another instance of this widget, a monitoring script, ... - stays in
+    sync live.
+
+    Args:
+        schedule_name: a stable, user-chosen name for the schedule. Unlike
+            a widget's `gui_id` (regenerated every time the widget is
+            constructed), this name is what lets a widget that was closed
+            and reopened find its own previously persisted schedule again.
+    """
+    return EndpointInfo(
+        endpoint=f"{EndpointType.USER.value}/schedule_widget/schedule/{schedule_name}",
+        message_type=VariableMessage,
+        message_op=MessageOp.SET_PUBLISH,
+    )

debye_bec/bec_widgets/widgets/scheduler/guard.py

@@ -0,0 +1,160 @@
+"""
+A small, self-contained utility for "if signal X drops below A, pause; once
+it's back above B, resume" (hysteresis) behavior - e.g. auto-pausing scans
+while the beam current is too low.
+
+This module knows nothing about schedules, scans, or the rest of this
+plugin. It only monitors one BEC device's live readback value
+(`MessageEndpoints.device_readback`, the same endpoint the device server
+publishes for every monitored device) and emits Qt signals when the value
+crosses one of the two configured thresholds. `schedule_widget.py` is the
+only place that connects those signals to schedule-specific behavior
+(aborting/deferring a scan item) - see its module docstring for how.
+"""
+
+from __future__ import annotations
+
+import threading
+
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
+from pydantic import BaseModel
+from qtpy.QtCore import QObject, Signal
+
+logger = bec_logger.logger
+
+
+class GuardSettings(BaseModel):
+    """Persisted configuration for one `SignalGuard`."""
+
+    enabled: bool = False
+    device_name: str | None = None
+    pause_below: float | None = None
+    resume_above: float | None = None
+
+
+class SignalGuard(QObject):
+    """
+    Monitors one device's readback value and reports hysteresis-based
+    pause/resume crossings.
+
+    - `paused` fires the first time the value drops below `pause_below`
+      (not again while it stays low).
+    - `resumed` fires the first time it then climbs back above
+      `resume_above` (not again while it stays high).
+    A value sitting between the two thresholds never re-triggers either
+    signal - that gap is the point of using two thresholds instead of one,
+    so a value oscillating right at a single cutoff wouldn't cause rapid
+    pause/resume flapping.
+
+    Callbacks from `RedisConnector.register()` run on a background
+    (Redis-listener) thread, not the Qt GUI thread; `paused`/`resumed` are
+    Qt signals, so connecting to them with the default (auto) connection
+    type safely marshals delivery onto whatever thread the receiver lives
+    on - no extra locking needed on the receiving end.
+    """
+
+    paused = Signal(float)
+    resumed = Signal(float)
+
+    def __init__(self, connector, parent=None):
+        super().__init__(parent)
+        self._connector = connector
+        self._lock = threading.Lock()
+        self._clear_event = threading.Event()
+        self._clear_event.set()  # not blocking until configured/proven otherwise
+
+        self.enabled = False
+        self.device_name: str | None = None
+        self.pause_below: float | None = None
+        self.resume_above: float | None = None
+        self.current_value: float | None = None
+
+        self._subscribed_endpoint = None
+
+    def configure(self, settings: GuardSettings):
+        """(Re)configure and (re)subscribe. Safe to call repeatedly, e.g. after editing settings."""
+        self._unsubscribe()
+
+        self.enabled = settings.enabled
+        self.device_name = settings.device_name
+        self.pause_below = settings.pause_below
+        self.resume_above = settings.resume_above
+        self.current_value = None
+        self._clear_event.set()
+
+        if self.enabled and self.device_name and self.pause_below is not None and self.resume_above is not None:
+            self._subscribed_endpoint = MessageEndpoints.device_readback(self.device_name)
+            self._connector.register(topics=self._subscribed_endpoint, cb=self._on_readback)
+
+    def _unsubscribe(self):
+        if self._subscribed_endpoint is not None:
+            try:
+                self._connector.unregister(topics=self._subscribed_endpoint, cb=self._on_readback)
+            except Exception:  # pylint: disable=broad-except
+                logger.exception("Failed to unsubscribe SignalGuard from %s", self._subscribed_endpoint)
+            self._subscribed_endpoint = None
+
+    def _on_readback(self, msg):
+        device_msg = getattr(msg, "value", None)
+        if device_msg is None:
+            return
+        value = _extract_value(device_msg, self.device_name)
+        if value is None:
+            return
+
+        crossed_pause = crossed_resume = False
+        with self._lock:
+            self.current_value = value
+            if self._clear_event.is_set():
+                if value < self.pause_below:
+                    self._clear_event.clear()
+                    crossed_pause = True
+            elif value > self.resume_above:
+                self._clear_event.set()
+                crossed_resume = True
+
+        if crossed_pause:
+            logger.info("SignalGuard: %s dropped to %s (below %s) - pausing.", self.device_name, value, self.pause_below)
+            self.paused.emit(value)
+        elif crossed_resume:
+            logger.info("SignalGuard: %s recovered to %s (above %s) - resuming.", self.device_name, value, self.resume_above)
+            self.resumed.emit(value)
+
+    def is_clear(self) -> bool:
+        """True if not currently blocking (disabled, or value above the resume threshold)."""
+        return (not self.enabled) or self._clear_event.is_set()
+
+    def wait_until_clear(self, should_abort, poll_interval: float = 0.5) -> bool:
+        """
+        Blocks the calling (non-GUI) thread until `is_clear()` becomes
+        True, checking `should_abort()` between polls so an operator abort
+        can still interrupt the wait. Returns True if it cleared, False if
+        `should_abort()` returned True first.
+        """
+        if self.is_clear():
+            return True
+        while not should_abort():
+            if self._clear_event.wait(timeout=poll_interval):
+                return True
+        return False
+
+    def cleanup(self):
+        self._unsubscribe()
+
+
+def _extract_value(device_msg, device_name: str | None):
+    """
+    `DeviceMessage.signals` is keyed by signal name, e.g. {"beam_current":
+    {"value": ..., "timestamp": ...}, ...}. For a simple scalar device the
+    primary signal is usually named after the device itself; fall back to
+    the first signal found if not (e.g. a differently-named primary signal
+    on a compound device).
+    """
+    signals = getattr(device_msg, "signals", None) or {}
+    if device_name in signals:
+        return signals[device_name].get("value")
+    for signal in signals.values():
+        if "value" in signal:
+            return signal["value"]
+    return None

debye_bec/bec_widgets/widgets/scheduler/guard_dialog.py

@@ -0,0 +1,94 @@
+"""Small settings dialog for `guard.GuardSettings` - kept separate from the guard logic itself."""
+
+from __future__ import annotations
+
+from qtpy.QtWidgets import (
+    QCheckBox,
+    QComboBox,
+    QDialog,
+    QDialogButtonBox,
+    QDoubleSpinBox,
+    QFormLayout,
+    QLabel,
+    QMessageBox,
+    QVBoxLayout,
+)
+
+from .guard import GuardSettings
+
+_DSPIN_RANGE = (-1e12, 1e12)
+
+
+class GuardSettingsDialog(QDialog):
+    """Configure (or disable) the auto-pause/resume guard for scan items."""
+
+    def __init__(self, settings: GuardSettings, device_names: list[str], parent=None):
+        super().__init__(parent)
+        self.setWindowTitle("Auto-pause on signal")
+
+        layout = QVBoxLayout(self)
+        info_label = QLabel(
+            "If enabled, the currently running scan is aborted as soon as the chosen "
+            "signal drops below the pause value, and automatically restarted once it "
+            "rises back above the resume value. Only scan items are affected - device "
+            "moves and custom/RPC commands are never interrupted by this."
+        )
+        info_label.setWordWrap(True)
+        layout.addWidget(info_label)
+
+        form = QFormLayout()
+        layout.addLayout(form)
+
+        self.enabled_check = QCheckBox("Enable auto-pause")
+        self.enabled_check.setChecked(settings.enabled)
+        form.addRow("", self.enabled_check)
+
+        self.device_combo = QComboBox()
+        self.device_combo.addItems(device_names)
+        if settings.device_name:
+            idx = self.device_combo.findText(settings.device_name)
+            if idx >= 0:
+                self.device_combo.setCurrentIndex(idx)
+        form.addRow("Signal (device)", self.device_combo)
+
+        self.pause_below_spin = QDoubleSpinBox()
+        self.pause_below_spin.setDecimals(4)
+        self.pause_below_spin.setRange(*_DSPIN_RANGE)
+        if settings.pause_below is not None:
+            self.pause_below_spin.setValue(settings.pause_below)
+        form.addRow("Pause below", self.pause_below_spin)
+
+        self.resume_above_spin = QDoubleSpinBox()
+        self.resume_above_spin.setDecimals(4)
+        self.resume_above_spin.setRange(*_DSPIN_RANGE)
+        if settings.resume_above is not None:
+            self.resume_above_spin.setValue(settings.resume_above)
+        form.addRow("Resume above", self.resume_above_spin)
+
+        buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
+        buttons.accepted.connect(self._on_accept)
+        buttons.rejected.connect(self.reject)
+        layout.addWidget(buttons)
+
+    def _on_accept(self):
+        if self.enabled_check.isChecked():
+            if not self.device_combo.currentText():
+                QMessageBox.warning(self, "Missing input", "Select a device to monitor.")
+                return
+            if self.resume_above_spin.value() <= self.pause_below_spin.value():
+                QMessageBox.warning(
+                    self,
+                    "Invalid thresholds",
+                    "'Resume above' must be greater than 'Pause below' (hysteresis gap).",
+                )
+                return
+        self.accept()
+
+    def result_settings(self) -> GuardSettings:
+        """Valid after `exec_()` returns `QDialog.Accepted`."""
+        return GuardSettings(
+            enabled=self.enabled_check.isChecked(),
+            device_name=self.device_combo.currentText() or None,
+            pause_below=self.pause_below_spin.value(),
+            resume_above=self.resume_above_spin.value(),
+        )

debye_bec/bec_widgets/widgets/scheduler/item_dialog.py

@@ -0,0 +1,238 @@
+"""
+The dialog behind the schedule widget's "Add..."/"Edit..." buttons.
+
+Rather than asking the operator to remember and type
+`scans.xas_simple_scan(12000, 14000, 2, 10)`-style commands, this presents:
+
+- a "Scan" tab: BEC's own `bec_widgets` `ScanControl` widget, embedded
+  as-is - scan selection, its live-generated per-scan argument form,
+  docs tooltips, metadata, "recall last scan parameters", all of it.
+  Reusing it instead of a plugin-owned reimplementation means this stays
+  in sync with BEC's scan capabilities for free, and looks/behaves exactly
+  like the scan controls an operator already knows from elsewhere in the
+  GUI. `ScanControl.button_run_scan` ("Start") is hidden here: this dialog
+  only ever wants the configured scan name/args/kwargs, never an
+  immediate submission - see `_collect_scan_result`.
+- a "Move" tab: pick a device and a target value/relative flag;
+- a "Custom" tab: a free-text field, for anything else (including RPC
+  calls to other widgets), evaluated the same way as the other tabs.
+
+Whichever tab is used, the dialog's only output is the same kind of plain
+command string the executor already knows how to run - this dialog adds a
+friendlier way to *build* that string, it doesn't change what happens with
+it afterwards. `kind`/`form_state` are carried along purely so "Edit..."
+can reopen the dialog pre-filled instead of asking the user to start over.
+"""
+
+from __future__ import annotations
+
+from bec_widgets.widgets.control.scan_control.scan_control import ScanControl, ScanParameterConfig
+from qtpy.QtWidgets import (
+    QCheckBox,
+    QComboBox,
+    QDialog,
+    QDialogButtonBox,
+    QDoubleSpinBox,
+    QFormLayout,
+    QLabel,
+    QLineEdit,
+    QMessageBox,
+    QTabWidget,
+    QVBoxLayout,
+    QWidget,
+)
+
+from .scan_form import list_movable_device_names
+
+_DSPIN_RANGE = (-1e12, 1e12)
+
+
+class ScheduleItemDialog(QDialog):
+    """Add or edit one schedule item, via ScanControl, a move form, or free text."""
+
+    def __init__(self, scans, dev, parent=None, initial: dict | None = None, client=None):
+        super().__init__(parent)
+        self.setWindowTitle("Schedule item")
+        self.setMinimumSize(520, 480)
+
+        self._scans = scans
+        self._dev = dev
+        self._client = client
+
+        layout = QVBoxLayout(self)
+        self.tabs = QTabWidget()
+        layout.addWidget(self.tabs)
+
+        self._build_scan_tab()
+        self._build_move_tab()
+        self._build_custom_tab()
+
+        buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
+        buttons.accepted.connect(self._on_accept)
+        buttons.rejected.connect(self.reject)
+        layout.addWidget(buttons)
+
+        self._apply_initial(initial or {})
+
+    # ------------------------------------------------------------------ #
+    # Scan tab - embeds BEC's own ScanControl widget
+    # ------------------------------------------------------------------ #
+    def _build_scan_tab(self):
+        tab = QWidget()
+        layout = QVBoxLayout(tab)
+
+        # client=None resolves to the same process-wide BEC client
+        # (bec_dispatcher.client) our own widget uses - no second Redis
+        # connection is opened.
+        self.scan_control = ScanControl(parent=tab, client=self._client)
+        self.scan_control.button_run_scan.hide()
+        layout.addWidget(self.scan_control)
+
+        hint = QLabel(
+            "Configure the scan above, then confirm with OK below - it will be "
+            "added to the schedule, not started immediately."
+        )
+        hint.setWordWrap(True)
+        hint.setStyleSheet("color: gray;")
+        layout.addWidget(hint)
+
+        self.tabs.addTab(tab, "Scan")
+
+    def _collect_scan_result(self) -> dict:
+        # Same call ScanControl.run_scan() makes before actually
+        # submitting, to resolve a typed-but-unconfirmed scan name.
+        self.scan_control.validate_scan_selection()
+        scan_name = self.scan_control.current_scan
+        if not scan_name:
+            raise ValueError("No scan selected.")
+
+        # bec_object=False: plain, repr-able values (e.g. a device name
+        # string rather than the live DeviceBase instance) - needed since
+        # the result has to survive a round-trip through Redis as text and
+        # be re-evaluated later, not just used in-process immediately.
+        args, kwargs = self.scan_control.get_scan_parameters(bec_object=False)
+        command = _format_scan_call(scan_name, args, kwargs)
+        return {
+            "command": command,
+            "kind": "scan",
+            "form_state": {"scan_name": scan_name, "args": args, "kwargs": kwargs},
+        }
+
+    def _prefill_scan_tab(self, scan_name: str, args: list, kwargs: dict):
+        # ScanControl restores parameters for a scan from its own config
+        # cache (see `ScanControl.restore_scan_parameters`); pre-loading
+        # that cache before switching to the scan reuses that mechanism
+        # instead of poking at its internal argument widgets directly.
+        self.scan_control.config.scans[scan_name] = ScanParameterConfig(
+            name=scan_name, args=args, kwargs=kwargs
+        )
+        self.scan_control.current_scan = scan_name
+
+    # ------------------------------------------------------------------ #
+    # Move tab
+    # ------------------------------------------------------------------ #
+    def _build_move_tab(self):
+        tab = QWidget()
+        form = QFormLayout(tab)
+
+        self.move_device_combo = QComboBox()
+        self.move_device_combo.addItems(list_movable_device_names(self._dev))
+        form.addRow("Device", self.move_device_combo)
+
+        self.move_value_spin = QDoubleSpinBox()
+        self.move_value_spin.setDecimals(6)
+        self.move_value_spin.setRange(*_DSPIN_RANGE)
+        form.addRow("Target value", self.move_value_spin)
+
+        self.move_relative_check = QCheckBox("Relative move")
+        form.addRow("", self.move_relative_check)
+
+        self.tabs.addTab(tab, "Move")
+
+    def _collect_move_result(self) -> dict:
+        device_name = self.move_device_combo.currentText()
+        if not device_name:
+            raise ValueError("No movable device available/selected.")
+        value = self.move_value_spin.value()
+        relative = self.move_relative_check.isChecked()
+        return {
+            "command": f"dev.{device_name}.move({value!r}, relative={relative!r})",
+            "kind": "move",
+            "form_state": {"device_name": device_name, "value": value, "relative": relative},
+        }
+
+    # ------------------------------------------------------------------ #
+    # Custom tab
+    # ------------------------------------------------------------------ #
+    def _build_custom_tab(self):
+        tab = QWidget()
+        layout = QVBoxLayout(tab)
+        layout.addWidget(
+            QLabel(
+                "Free-form command, evaluated against `scans` and `dev` - use this for "
+                "anything the Scan/Move tabs don't cover, e.g. an RPC call to another widget."
+            )
+        )
+        self.custom_edit = QLineEdit()
+        self.custom_edit.setPlaceholderText("scans.xas_simple_scan(12000, 14000, 2, 10)")
+        layout.addWidget(self.custom_edit)
+        layout.addStretch(1)
+        self.tabs.addTab(tab, "Custom")
+
+    def _collect_custom_result(self) -> dict:
+        text = self.custom_edit.text().strip()
+        if not text:
+            raise ValueError("Command must not be empty.")
+        return {"command": text, "kind": "custom", "form_state": {"text": text}}
+
+    # ------------------------------------------------------------------ #
+    # pre-fill (edit mode) / result extraction
+    # ------------------------------------------------------------------ #
+    def _apply_initial(self, initial: dict):
+        kind = initial.get("kind")
+        state = initial.get("form_state") or {}
+
+        if kind == "scan" and state.get("scan_name"):
+            self._prefill_scan_tab(
+                state["scan_name"], state.get("args") or [], state.get("kwargs") or {}
+            )
+            self.tabs.setCurrentIndex(0)
+        elif kind == "move" and state.get("device_name"):
+            idx = self.move_device_combo.findText(state["device_name"])
+            if idx >= 0:
+                self.move_device_combo.setCurrentIndex(idx)
+            self.move_value_spin.setValue(float(state.get("value", 0.0)))
+            self.move_relative_check.setChecked(bool(state.get("relative", False)))
+            self.tabs.setCurrentIndex(1)
+        elif initial.get("command"):
+            # "custom" kind, or a legacy/unrecognized item - fall back to
+            # showing the raw command text as-is.
+            self.custom_edit.setText(state.get("text", initial["command"]))
+            self.tabs.setCurrentIndex(2)
+
+    def _on_accept(self):
+        try:
+            result = self._collect_result()
+        except ValueError as exc:
+            QMessageBox.warning(self, "Missing input", str(exc))
+            return
+        self._result = result
+        self.accept()
+
+    def _collect_result(self) -> dict:
+        current = self.tabs.currentIndex()
+        if current == 0:
+            return self._collect_scan_result()
+        if current == 1:
+            return self._collect_move_result()
+        return self._collect_custom_result()
+
+    def result(self) -> dict:
+        """Valid after `exec_()` returns `QDialog.Accepted`."""
+        return self._result
+
+
+def _format_scan_call(scan_name: str, args: list, kwargs: dict) -> str:
+    parts = [repr(a) for a in args]
+    parts += [f"{name}={value!r}" for name, value in kwargs.items()]
+    return f"scans.{scan_name}({', '.join(parts)})"

debye_bec/bec_widgets/widgets/scheduler/notification_dialog.py

@@ -0,0 +1,59 @@
+"""Small settings dialog for `notifications.NotificationSettings` - kept separate from the send logic itself."""
+
+from __future__ import annotations
+
+from qtpy.QtWidgets import QCheckBox, QDialog, QDialogButtonBox, QLabel, QVBoxLayout
+
+from .notifications import NotificationSettings
+
+
+class NotificationSettingsDialog(QDialog):
+    """Choose whether, and for which item types, finish/fail notifications are sent."""
+
+    def __init__(self, settings: NotificationSettings, parent=None):
+        super().__init__(parent)
+        self.setWindowTitle("Notifications")
+
+        layout = QVBoxLayout(self)
+        info_label = QLabel(
+            "Sends a BEC notification (see MessageEndpoints.notification) whenever a "
+            "schedule item finishes or fails. Whether it reaches you as a message depends "
+            "on your facility's notification routing (SciHub + Signal/Teams/SciLog) being "
+            "configured for these events - ask your beamline admin if you're not sure."
+        )
+        info_label.setWordWrap(True)
+        layout.addWidget(info_label)
+
+        self.enabled_check = QCheckBox("Send notifications")
+        self.enabled_check.setChecked(settings.enabled)
+        layout.addWidget(self.enabled_check)
+
+        self.scan_check = QCheckBox("Scans")
+        self.scan_check.setChecked(settings.notify_scan)
+        layout.addWidget(self.scan_check)
+
+        self.move_check = QCheckBox("Movements")
+        self.move_check.setChecked(settings.notify_move)
+        layout.addWidget(self.move_check)
+
+        self.rpc_check = QCheckBox("RPC / custom commands")
+        self.rpc_check.setChecked(settings.notify_rpc)
+        layout.addWidget(self.rpc_check)
+
+        for check in (self.scan_check, self.move_check, self.rpc_check):
+            self.enabled_check.toggled.connect(check.setEnabled)
+            check.setEnabled(settings.enabled)
+
+        buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
+        buttons.accepted.connect(self.accept)
+        buttons.rejected.connect(self.reject)
+        layout.addWidget(buttons)
+
+    def result_settings(self) -> NotificationSettings:
+        """Valid after `exec_()` returns `QDialog.Accepted`."""
+        return NotificationSettings(
+            enabled=self.enabled_check.isChecked(),
+            notify_scan=self.scan_check.isChecked(),
+            notify_move=self.move_check.isChecked(),
+            notify_rpc=self.rpc_check.isChecked(),
+        )

debye_bec/bec_widgets/widgets/scheduler/notifications.py

@@ -0,0 +1,94 @@
+"""
+A small, self-contained "tell the user when something finished/failed"
+feature, independent of scheduling/execution logic.
+
+Sends via BEC's own notification transport: `MessageEndpoints.notification`
+carries a `messages.NotificationMessage`, the same message type/endpoint
+BEC's own scan lifecycle events (e.g. "scan_completed") use. Reusing it
+means schedule-item notifications are just more entries in the same
+mechanism, not a plugin-invented side channel - and if your facility runs
+SciHub with messaging-service routing configured (Signal/Teams/SciLog),
+they can be forwarded out the same way any other BEC notification would
+be, by adding a route for the event names below in that routing config.
+
+Nothing here is scheduling/BECWidget-specific: `notify_item_finished` only
+needs a connector, a settings object and a few plain values, so it can be
+called from `schedule_widget.py` without either module depending on the
+other's internals.
+"""
+
+from __future__ import annotations
+
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
+from bec_lib.messages import MessagingServiceTextContent, NotificationMessage
+from pydantic import BaseModel
+
+logger = bec_logger.logger
+
+# Namespaced so they can't collide with BEC's own built-in event names
+# (e.g. "scan_completed"); an admin who wants these routed to Signal/Teams/
+# SciLog can add a route for exactly these names in their notification
+# routing config (`MessageEndpoints.notification_config()`).
+_EVENT_NAMES = {
+    ("scan", True): "schedule_scan_completed",
+    ("scan", False): "schedule_scan_failed",
+    ("move", True): "schedule_move_completed",
+    ("move", False): "schedule_move_failed",
+    ("custom", True): "schedule_rpc_completed",
+    ("custom", False): "schedule_rpc_failed",
+}
+
+
+class NotificationSettings(BaseModel):
+    """Persisted per-schedule notification preferences."""
+
+    enabled: bool = False
+    notify_scan: bool = True
+    notify_move: bool = True
+    # "custom" is what this plugin calls the free-text tab (see
+    # item_dialog.py) - it covers RPC calls to other widgets as well as
+    # any hand-typed command, so it's exposed to the user as "RPC".
+    notify_rpc: bool = True
+
+
+def _enabled_for_kind(settings: NotificationSettings, kind: str) -> bool:
+    if not settings.enabled:
+        return False
+    return {
+        "scan": settings.notify_scan,
+        "move": settings.notify_move,
+        "custom": settings.notify_rpc,
+    }.get(kind, False)
+
+
+def notify_item_finished(
+    connector,
+    schedule_name: str,
+    kind: str,
+    command: str,
+    success: bool,
+    settings: NotificationSettings,
+    error: str | None = None,
+):
+    """
+    Sends a `NotificationMessage` if notifications are enabled for `kind`.
+    Never raises - a notification failing to send should not affect
+    schedule execution.
+    """
+    if not _enabled_for_kind(settings, kind):
+        return
+
+    event = _EVENT_NAMES.get((kind, success), "schedule_item_finished")
+    status_word = "completed" if success else "failed"
+    text = f"[{schedule_name}] {kind} {status_word}: {command}"
+    if error:
+        text += f"\n{error.strip().splitlines()[-1]}"
+
+    try:
+        connector.send(
+            MessageEndpoints.notification(event),
+            NotificationMessage(event=event, message=[MessagingServiceTextContent(content=text)]),
+        )
+    except Exception:  # pylint: disable=broad-except
+        logger.exception("Failed to send schedule notification for event '%s'", event)

debye_bec/bec_widgets/widgets/scheduler/register_scheduler.py

@@ -0,0 +1,15 @@
+def main():  # pragma: no cover
+    from qtpy import PYSIDE6
+
+    if not PYSIDE6:
+        print("PYSIDE6 is not available in the environment. Cannot patch designer.")
+        return
+    from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
+
+    from .scheduler_plugin import SchedulerPlugin
+
+    QPyDesignerCustomWidgetCollection.addCustomWidget(SchedulerPlugin())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

debye_bec/bec_widgets/widgets/scheduler/scan_form.py

@@ -0,0 +1,16 @@
+"""
+Small device-listing helper for the "Move" tab of `item_dialog.py`.
+
+Scan parameter introspection/forms used to live in this module too, but
+that's now handled by embedding BEC's own `bec_widgets` `ScanControl`
+widget directly (see `item_dialog.py`) instead of a plugin-owned
+reimplementation - one less UI to keep in sync with BEC's scan
+capabilities.
+"""
+
+from __future__ import annotations
+
+
+def list_movable_device_names(dev) -> list[str]:
+    """Device names that expose `.move(...)` (i.e. positioner-like devices)."""
+    return sorted(name for name, device in dev.items() if hasattr(device, "move"))

debye_bec/bec_widgets/widgets/scheduler/schedule_item.py

@@ -0,0 +1,76 @@
+"""
+Plain (non-BECMessage) data model for one schedule entry and the schedule
+as a whole.
+
+These are ordinary pydantic models used only client-side, for validation
+and convenience. They are never sent over Redis as their own type - only
+ever as the `.value` payload of a `bec_lib.messages.VariableMessage` (see
+`endpoints.schedule` for why).
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from .guard import GuardSettings
+from .notifications import NotificationSettings
+
+
+class ScheduleItemStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    ABORTED = "aborted"
+
+
+# Which tab of the Add/Edit dialog built `command`, and therefore how to
+# re-open that dialog pre-filled with the same values for editing. "custom"
+# covers hand-typed text (including RPC calls to other widgets), which has
+# no structured `form_state` to restore beyond the raw text itself.
+ScheduleItemKind = Literal["scan", "move", "custom"]
+
+
+class ScheduleItem(BaseModel):
+    """A single command in the schedule, plus its execution bookkeeping."""
+
+    item_id: str
+    command: str
+    status: ScheduleItemStatus = ScheduleItemStatus.PENDING
+
+    # How `command` was produced, and (for "scan"/"move") the structured
+    # inputs behind it, so the Add/Edit dialog can be reopened pre-filled
+    # instead of asking the user to re-type everything. `command` itself
+    # remains the single source of truth actually evaluated at execution
+    # time - `kind`/`form_state` only drive the UI.
+    kind: ScheduleItemKind = "custom"
+    form_state: dict | None = None
+
+    # Bookkeeping used to reconnect to a submission that is still (or was)
+    # in flight on the BEC scan/device server, after this widget has been
+    # closed and reopened.
+    request_id: str | None = None
+    scan_id: str | None = None
+    scan_number: int | None = None
+
+    error: str | None = None
+    started_at: float | None = None
+    finished_at: float | None = None
+
+
+class Schedule(BaseModel):
+    """The full, persisted state of one schedule-widget instance."""
+
+    schedule_name: str
+    items: list[ScheduleItem] = Field(default_factory=list)
+    is_running: bool = False
+
+    # Settings for two independent, optional features (see notifications.py
+    # and guard.py) - persisted here alongside the schedule itself so they
+    # survive a widget restart too, but their *logic* lives entirely in
+    # those separate modules; this is just where their settings are stored.
+    notifications: NotificationSettings = Field(default_factory=NotificationSettings)
+    guard: GuardSettings = Field(default_factory=GuardSettings)

debye_bec/bec_widgets/widgets/scheduler/schedule_logic.py

@@ -0,0 +1,60 @@
+"""
+Pure bookkeeping over a list of `ScheduleItem`s - no Qt, no locking, no I/O.
+Kept separate from `schedule_widget.py` for the same reason `guard.py` and
+`notifications.py` are: it's testable on its own, and the widget should
+only be responsible for Qt/orchestration, not figuring out which item runs
+next.
+
+The one invariant everything here assumes and preserves: at any moment, a
+schedule's items form a (possibly empty) prefix that is no longer PENDING
+(RUNNING - at most one, the one currently executing - COMPLETED, FAILED or
+ABORTED), followed by a suffix that is entirely PENDING. `ScheduleWidget`
+is responsible for only ever mutating the PENDING suffix (see its module
+docstring) and for calling these functions while holding its lock; nothing
+here does its own locking.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from .schedule_item import ScheduleItem, ScheduleItemStatus
+
+
+def protected_prefix_length(items: list[ScheduleItem]) -> int:
+    """How many items, from the start, are no longer PENDING."""
+    count = 0
+    for item in items:
+        if item.status != ScheduleItemStatus.PENDING:
+            count += 1
+        else:
+            break
+    return count
+
+
+def index_of(items: list[ScheduleItem], item_id: str | None) -> int | None:
+    if item_id is None:
+        return None
+    for i, item in enumerate(items):
+        if item.item_id == item_id:
+            return i
+    return None
+
+
+def pick_next_runnable(items: list[ScheduleItem]) -> ScheduleItem | Literal["stop"] | None:
+    """
+    What the execution loop should do next: always re-derived from scratch
+    (never a remembered index/object) so edits made to the PENDING suffix
+    between calls are picked up correctly.
+
+    Returns the next item to run/attach to, the string `"stop"` if an
+    earlier item failed/was aborted (execution stays parked there until the
+    operator intervenes), or `None` if every item is COMPLETED.
+    """
+    for item in items:
+        if item.status == ScheduleItemStatus.COMPLETED:
+            continue
+        if item.status in (ScheduleItemStatus.FAILED, ScheduleItemStatus.ABORTED):
+            return "stop"
+        return item  # PENDING, or RUNNING (reconciled as still active)
+    return None

debye_bec/bec_widgets/widgets/scheduler/scheduler.py

@@ -0,0 +1,995 @@
+"""
+ScheduleWidget - persists an ordered schedule of BEC commands (scans,
+device moves, ...) to Redis, executes them one by one, and survives being
+closed and reopened - including while an item is still running
+server-side. Items that have not started yet can be added, reordered or
+deleted while the schedule is executing; items that have already started
+or finished cannot.
+
+Persistence
+-----------
+The whole schedule (list of items + status) is stored as one
+`VariableMessage` document under a stable, named endpoint
+(`endpoints.schedule(schedule_name)`), re-written and republished on every
+change. See `endpoints.py` for the reasoning behind that design.
+
+Building a command
+-------------------
+The "Add..."/"Edit..." buttons open `item_dialog.ScheduleItemDialog`,
+which lets the operator pick a registered scan through BEC's own
+`bec_widgets` `ScanControl` widget (embedded as-is, with its "Start"
+button hidden so it only ever configures a scan instead of submitting
+one) or a device move from a small form, instead of typing a command by
+hand - while still allowing every parameter value to be changed. A
+"Custom" tab keeps a free-text command for anything else (e.g. an RPC
+call to another widget). Whichever tab is used, the result is the same
+kind of plain command string described below - the dialog only changes
+how that string gets built.
+
+Execution model
+----------------
+Each command (e.g. "scans.xas_simple_scan(12000, 14000, 2, 10)" or
+"dev.samx.move(12.3, relative=True)") is evaluated against a small,
+fixed namespace of `{"scans": self.scans, "dev": self.dev}`. This mirrors
+how `bec_lib.script_executor` runs persisted script text: it does not
+invent a second way to talk to devices/scans, it drives the exact same
+client objects an operator would use interactively.
+
+Both `scans.xxx(...)` and `dev.<name>.move(...)` (which internally calls
+`scans.mv(...)`) submit *asynchronously* to BEC's scan queue and return a
+`ScanReport` immediately, with `.request.requestID`, `.wait()` and
+`.cancel()`. Submission already happened server-side by the time `.wait()`
+returns - closing this widget (or even the whole GUI process) does not
+stop a scan or move that has already been submitted; it keeps running on
+the scan/device server.
+
+Reconciliation after being closed and reopened
+------------------------------------------------
+If this widget is closed while an item is RUNNING and reopened later, we
+must not blindly resume from "pending" (which could re-submit a scan that
+is still running) nor blindly trust our last persisted status forever
+(the scan may have long since finished). Instead, on load we ask BEC's
+*own*, live queue state: `client.queue.queue_storage`, freshly seeded from
+`MessageEndpoints.scan_queue_status()`, can look up any earlier
+`request_id` via `find_queue_item_by_requestID()`. If it's still
+pending/running there, we re-attach and wait; if it's no longer in the
+queue, it has left the system (finished, successfully or not) while we
+were disconnected, and we mark it completed and move on.
+
+Editing while running - the "protected prefix" invariant
+------------------------------------------------------------
+`run_schedule()` walks the items strictly in order in a background task
+(`BECConnector.submit_task`). To let the operator add/reorder/delete
+items that haven't started yet *while that task is running*, without
+racing it, every mutation goes through `self._lock` and respects one
+invariant: at any moment, the schedule's items form a (possibly empty)
+prefix of items that are no longer PENDING (RUNNING - at most one, the
+one currently executing - COMPLETED/FAILED/ABORTED), followed by a suffix
+of items that are all still PENDING. The pure bookkeeping over that
+invariant (`protected_prefix_length`, `index_of`, `pick_next_runnable`)
+lives in `schedule_logic.py`, kept separate the same way `guard.py` and
+`notifications.py` are - no Qt, no locking, independently testable; this
+widget only calls it while holding `self._lock`. Edits are only ever
+allowed to touch the PENDING suffix:
+  - `add_item(..., index=...)` clamps `index` up to at least the prefix
+    length, so a new item can never be inserted before something already
+    running or finished.
+  - `move_item`/`delete_item` refuse (raise `RuntimeError`) if the target
+    item's status isn't PENDING.
+The execution task itself never keeps a stale index across its
+(potentially long) `report.wait()` calls - each time it needs the next
+item, it re-scans `self.schedule.items` from the start under the lock, so
+edits made to the PENDING suffix while an earlier item is executing are
+picked up correctly.
+
+Multi-instance note: `_on_remote_update` (another widget instance, or a
+script, publishing a new version of this schedule) replaces
+`self.schedule` wholesale. If *this* instance is currently running its
+own execution task, applying a wholesale replacement here would orphan
+the `ScheduleItem` object that task is watching, so incoming remote
+updates are ignored while `self.schedule.is_running` locally - this
+instance's own state wins until it finishes running, at which point its
+next `_persist()` call re-publishes the authoritative state anyway. This
+is a simplification: true multi-writer merging (e.g. concurrently editing
+the PENDING suffix from two open widgets while a third is executing it)
+is out of scope here.
+
+Notes / things to adapt for your BEC version and deployment
+-------------------------------------------------------------
+- `eval()` is used deliberately for the "Custom" tab and to actually run
+  every item, with an empty `__builtins__` and a namespace restricted to
+  `scans`/`dev`, matching BEC's own trust model for script text (see
+  `bec_lib.script_executor`). The Scan/Move tabs never eval user text -
+  values come from typed Qt widgets and are placed into the generated
+  command with `repr()`.
+- `QueueItem.status` is a plain string on the installed bec_lib version
+  this was written against; the exact set of literal values was not
+  fully pinned down against a live scan server, so
+  `_is_queue_item_active()` treats anything containing "PENDING" or
+  "RUNNING" as still active and everything else as finished.
+
+Two independent, optional features
+-------------------------------------
+Both are implemented in their own modules with no dependency on
+scheduling/execution internals - `schedule_widget.py` only wires their
+signals/functions into the execution loop:
+
+- **Notifications** (`notifications.py`/`notification_dialog.py`): sends a
+  BEC `NotificationMessage` whenever an item finishes or fails, if enabled
+  for that item's kind (scan/move/rpc, independently toggleable). Fires
+  from `_execute_item()` after an item's final status is known.
+- **Auto-pause guard** (`guard.py`/`guard_dialog.py`): a `SignalGuard`
+  monitors one device's live readback and, on hysteresis-based
+  pause/resume crossings, emits Qt signals `schedule_widget.py` reacts to.
+  Applies to scan items only - see "Guard interaction" below.
+
+Guard interaction with execution
+-----------------------------------
+When the guard is enabled and about to start a `kind == "scan"` item,
+`_execute_item` first calls `self._guard.wait_until_clear(...)`, blocking
+(without marking the item RUNNING, so it stays PENDING and visibly
+unaffected by the "protected prefix" invariant) until the signal is back
+above the resume threshold or the operator aborts. If the guard trips
+*while* a scan item is RUNNING, `_on_guard_paused` cancels its
+`ScanReport` (the same mechanism `abort_schedule()` uses) and sets
+`self._guard_interrupt_requested`; `_execute_item`'s exception handler
+checks that flag and, if set, resets the item back to PENDING (clearing
+its request/scan bookkeeping) instead of marking it ABORTED - so the
+normal execution loop simply picks it up again once the guard clears,
+with no special "resume" code path needed. Device moves and custom/RPC
+items are never touched by the guard (checked via `item.kind == "scan"`
+in both places) - note this means a scan typed into the free-text
+"Custom" tab is *not* guard-protected, since we can't tell hand-typed
+text apart from a scan call; use the "Scan" tab if you need that. One
+more limitation: a scan reconciled as "still running from a previous
+session" (see "Reconciliation" above) is only *waited on* via
+`_await_running_item`, not submitted through `_execute_item`, so there is
+no live `ScanReport` for the guard to cancel if it trips during that
+narrow window - it will simply be picked up again once that scan finishes
+on its own.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+import traceback
+import uuid
+
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
+from bec_lib.messages import VariableMessage
+from bec_widgets.utils.bec_connector import ConnectionConfig
+from bec_widgets.utils.bec_widget import BECWidget
+from bec_widgets.utils.error_popups import SafeSlot
+from qtpy.QtCore import Qt, Signal
+from qtpy.QtWidgets import (
+    QDialog,
+    QHBoxLayout,
+    QLabel,
+    QListWidget,
+    QListWidgetItem,
+    QMessageBox,
+    QPushButton,
+    QVBoxLayout,
+    QWidget,
+)
+
+from .endpoints import schedule as schedule_endpoint
+from .guard import GuardSettings, SignalGuard
+from .guard_dialog import GuardSettingsDialog
+from .item_dialog import ScheduleItemDialog
+from .notification_dialog import NotificationSettingsDialog
+from .notifications import NotificationSettings, notify_item_finished
+from .schedule_item import Schedule, ScheduleItem, ScheduleItemStatus
+from .schedule_logic import index_of, pick_next_runnable, protected_prefix_length
+
+logger = bec_logger.logger
+
+# The static, hard-coded example schedule requested for this plugin. A
+# real deployment would more likely let the user build this list purely
+# through the UI - the persistence/execution/reconciliation logic below
+# does not care where the commands came from.
+DEFAULT_COMMANDS = [
+    "scans.xas_simple_scan(12000, 14000, 2, 10)",
+    "dev.samx.move(12.3, relative=True)",
+    "scans.xas_simple_scan(13000, 17000, 1, 60)",
+]
+
+_ACTIVE_QUEUE_STATES = ("PENDING", "RUNNING")
+_ITEM_ID_ROLE = Qt.UserRole + 1  # QListWidgetItem data role used to map a row back to an item_id
+
+_STATUS_ICON = {
+    ScheduleItemStatus.PENDING: "\u23f3",  # hourglass
+    ScheduleItemStatus.RUNNING: "\u25b6",  # play
+    ScheduleItemStatus.COMPLETED: "\u2705",  # check mark
+    ScheduleItemStatus.FAILED: "\u274c",  # cross mark
+    ScheduleItemStatus.ABORTED: "\u23f9",  # stop
+}
+
+
+class ScheduleWidgetConfig(ConnectionConfig):
+    """
+    Per-instance identity/settings for `ScheduleWidget`, following the same
+    pattern `bec_widgets` itself uses (e.g. `ScanControlConfig`): a
+    `ConnectionConfig` subclass constructed in `__init__` when none is
+    passed, carrying whatever small settings distinguish one instance from
+    another.
+
+    `schedule_name` lives here rather than as a bare constructor kwarg for
+    that consistency - but note this is *not* where the schedule's actual
+    contents (items, guard/notification settings) live: `self.config` is
+    used for RPC identity and optional explicit GUI-layout save/restore,
+    not auto-persisted across an ordinary widget close/reopen the way this
+    plugin's Redis-backed `Schedule` document is. Conflating the two would
+    quietly reintroduce the "closed widget loses its schedule" problem this
+    plugin exists to solve.
+    """
+
+    schedule_name: str = "default_schedule"
+
+
+class Scheduler(BECWidget, QWidget):
+    """Schedule, persist and execute a sequence of BEC scan/device commands."""
+
+    ICON_NAME = "assignment_add"
+    USER_ACCESS = [
+        "run_schedule",
+        "abort_schedule",
+        "reset_schedule",
+        "get_status",
+        "add_item",
+        "edit_item",
+        "delete_item",
+        "move_item",
+        "move_item_up",
+        "move_item_down",
+    ]
+
+    # emitted (thread-safe, may come from the background execution task)
+    # whenever the schedule state changes and the UI should redraw
+    schedule_changed = Signal()
+
+    def __init__(
+        self,
+        parent=None,
+        client=None,
+        config: ScheduleWidgetConfig | dict | None = None,
+        gui_id: str | None = None,
+        schedule_name: str | None = None,
+        **kwargs,
+    ):
+        if config is None:
+            config = ScheduleWidgetConfig(
+                widget_class=self.__class__.__name__,
+                schedule_name=schedule_name or "default_schedule",
+            )
+        elif schedule_name is not None:
+            # Allow overriding even when a config object/dict was supplied,
+            # mirroring how ScanControl treats its own optional constructor
+            # args as overrides onto whatever config ends up in self.config.
+            if isinstance(config, dict):
+                config = dict(config, schedule_name=schedule_name)
+            else:
+                config.schedule_name = schedule_name
+        super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
+
+        # A stable name, NOT self.gui_id: gui_id is regenerated every time
+        # the widget is constructed, so it could never be used to find a
+        # previously persisted schedule again after the widget is reopened.
+        self.schedule_name = self.config.schedule_name
+
+        self.get_bec_shortcuts()  # -> self.client, self.dev, self.scans, self.queue
+        self.connector = self.client.connector
+
+        self._endpoint = schedule_endpoint(self.schedule_name)
+        self._lock = threading.Lock()  # guards self.schedule (see module docstring)
+        self._abort_requested = False
+        self._guard_interrupt_requested = False  # see "Guard interaction" in module docstring
+        self._current_report = None  # ScanReport of the item currently executing, if any
+        self._current_item_kind: str | None = None  # kind of the item _current_report belongs to
+        self._selected_item_id: str | None = None  # survives list repopulation on refresh
+
+        self.schedule_changed.connect(self._refresh_ui)
+
+        self.schedule: Schedule = self._load_or_seed_schedule()
+        self._reconcile_with_live_state()
+
+        self._guard = SignalGuard(self.connector, parent=self)
+        self._guard.paused.connect(self._on_guard_paused)
+        self._guard.resumed.connect(self._on_guard_resumed)
+        self._guard.configure(self.schedule.guard)
+
+        self._build_ui()
+        self._refresh_ui()
+
+        # Stay in sync with any other subscriber touching the same
+        # schedule (another instance of this widget, a script, ...).
+        self.bec_dispatcher.connect_slot(self._on_remote_update, self._endpoint)
+
+    # ------------------------------------------------------------------ #
+    # persistence
+    # ------------------------------------------------------------------ #
+    def _load_or_seed_schedule(self) -> Schedule:
+        logger.info("load or seed schedule")
+        msg: VariableMessage | None = self.connector.get(self._endpoint)
+        if msg is not None:
+            logger.info(f"Got msg from endpoint: {msg}")
+            return Schedule.model_validate(msg.value)
+
+        schedule = Schedule(
+            schedule_name=self.schedule_name,
+            items=[
+                ScheduleItem(item_id=str(uuid.uuid4()), command=cmd) for cmd in DEFAULT_COMMANDS
+            ],
+        )
+        self.connector.set_and_publish(
+            self._endpoint, VariableMessage(value=schedule.model_dump(mode="json"))
+        )
+        return schedule
+
+    def _persist_locked(self):
+        """Write `self.schedule` to Redis. Caller must hold `self._lock`."""
+        self.connector.set_and_publish(
+            self._endpoint, VariableMessage(value=self.schedule.model_dump(mode="json"))
+        )
+
+    def _persist(self):
+        with self._lock:
+            self._persist_locked()
+
+    @SafeSlot(dict, dict)
+    def _on_remote_update(self, msg_content: dict, metadata: dict):
+        value = msg_content.get("value") if msg_content else None
+        if not value or value.get("schedule_name") != self.schedule_name:
+            return
+        with self._lock:
+            if self.schedule.is_running:
+                # This instance's own execution task owns the current
+                # ScheduleItem objects (and will re-persist its own state
+                # again shortly); accepting a wholesale replacement here
+                # would orphan those objects. See module docstring.
+                logger.info(
+                    "Ignoring remote schedule update for '%s' while a local run is in progress.",
+                    self.schedule_name,
+                )
+                return
+            self.schedule = Schedule.model_validate(value)
+        self._apply_guard_settings()
+        self._refresh_ui()
+
+    # ------------------------------------------------------------------ #
+    # reconciliation: figure out real state after being closed/reopened
+    # ------------------------------------------------------------------ #
+    def _reconcile_with_live_state(self):
+        """
+        For any item last seen as RUNNING, ask BEC's own queue state -
+        not our local memory, which may be stale or from a previous
+        process - what actually happened to it while we were gone.
+        """
+        queue_storage = self.client.queue.queue_storage
+
+        # Force a fresh pull of the current queue snapshot from Redis
+        # rather than relying on pubsub messages we may have missed while
+        # this widget instance did not exist.
+        status_msg = self.connector.get(MessageEndpoints.scan_queue_status())
+        if status_msg is not None:
+            queue_storage.update_with_status(status_msg)
+
+        with self._lock:
+            changed = False
+            for item in self.schedule.items:
+                if item.status != ScheduleItemStatus.RUNNING or not item.request_id:
+                    continue
+                queue_item = queue_storage.find_queue_item_by_requestID(item.request_id)
+                if queue_item is not None and self._is_queue_item_active(queue_item):
+                    # genuinely still in flight server-side; leave it
+                    # RUNNING - run_schedule() will re-attach to it
+                    # instead of resubmitting.
+                    continue
+                item.status = ScheduleItemStatus.COMPLETED
+                item.finished_at = time.time()
+                changed = True
+            if changed:
+                self._persist_locked()
+
+    @staticmethod
+    def _is_queue_item_active(queue_item) -> bool:
+        return str(getattr(queue_item, "status", "")).upper() in _ACTIVE_QUEUE_STATES
+
+    # ------------------------------------------------------------------ #
+    # the "protected prefix" invariant (see module docstring) - the actual
+    # logic lives in schedule_logic.py, pure and independently testable;
+    # these thin wrappers just apply it to the current, lock-held state.
+    # ------------------------------------------------------------------ #
+    def _protected_prefix_length_locked(self) -> int:
+        return protected_prefix_length(self.schedule.items)
+
+    def _index_of_locked(self, item_id: str | None) -> int | None:
+        return index_of(self.schedule.items, item_id)
+
+    # ------------------------------------------------------------------ #
+    # UI
+    # ------------------------------------------------------------------ #
+    def _build_ui(self):
+        layout = QVBoxLayout(self)
+        layout.addWidget(QLabel(f"Schedule: {self.schedule_name}"))
+
+        self.list_widget = QListWidget()
+        self.list_widget.currentRowChanged.connect(self._on_selection_changed)
+        layout.addWidget(self.list_widget)
+
+        edit_row = QHBoxLayout()
+        self.add_btn = QPushButton("Add...")
+        self.edit_btn = QPushButton("Edit...")
+        self.delete_btn = QPushButton("Delete")
+        self.move_up_btn = QPushButton("Move Up")
+        self.move_down_btn = QPushButton("Move Down")
+        for button in (
+            self.add_btn,
+            self.edit_btn,
+            self.delete_btn,
+            self.move_up_btn,
+            self.move_down_btn,
+        ):
+            edit_row.addWidget(button)
+        layout.addLayout(edit_row)
+
+        self.add_btn.clicked.connect(self._on_add_clicked)
+        self.edit_btn.clicked.connect(self._on_edit_clicked)
+        self.delete_btn.clicked.connect(self._on_delete_clicked)
+        self.move_up_btn.clicked.connect(self._on_move_up_clicked)
+        self.move_down_btn.clicked.connect(self._on_move_down_clicked)
+
+        run_row = QHBoxLayout()
+        self.run_btn = QPushButton("Run / Continue")
+        self.abort_btn = QPushButton("Abort")
+        self.reset_btn = QPushButton("Reset")
+        for button in (self.run_btn, self.abort_btn, self.reset_btn):
+            run_row.addWidget(button)
+        layout.addLayout(run_row)
+
+        self.run_btn.clicked.connect(self.run_schedule)
+        self.abort_btn.clicked.connect(self.abort_schedule)
+        self.reset_btn.clicked.connect(self.reset_schedule)
+
+        settings_row = QHBoxLayout()
+        self.notifications_btn = QPushButton("Notifications...")
+        self.guard_btn = QPushButton("Auto-pause...")
+        settings_row.addWidget(self.notifications_btn)
+        settings_row.addWidget(self.guard_btn)
+        layout.addLayout(settings_row)
+
+        self.notifications_btn.clicked.connect(self._on_notifications_clicked)
+        self.guard_btn.clicked.connect(self._on_guard_clicked)
+
+        self.guard_status_label = QLabel()
+        self.guard_status_label.setStyleSheet("color: gray;")
+        layout.addWidget(self.guard_status_label)
+
+    @SafeSlot()
+    def _refresh_ui(self):
+        with self._lock:
+            items = list(self.schedule.items)
+            is_running = self.schedule.is_running
+
+        self.list_widget.blockSignals(True)
+        self.list_widget.clear()
+        selected_row = None
+        for row, item in enumerate(items):
+            text = f"{_STATUS_ICON.get(item.status, '?')}  {item.command}"
+            if item.error:
+                text += f"   ({item.error.strip().splitlines()[-1]})"
+            list_item = QListWidgetItem(text, self.list_widget)
+            list_item.setData(_ITEM_ID_ROLE, item.item_id)
+            if item.item_id == self._selected_item_id:
+                selected_row = row
+        if selected_row is not None:
+            self.list_widget.setCurrentRow(selected_row)
+        else:
+            self._selected_item_id = None
+        self.list_widget.blockSignals(False)
+
+        idx = None
+        selected_item = None
+        if self._selected_item_id is not None:
+            for i, item in enumerate(items):
+                if item.item_id == self._selected_item_id:
+                    idx, selected_item = i, item
+                    break
+
+        # Editable = the selected item hasn't started yet. Thanks to the
+        # protected-prefix invariant, everything after the first PENDING
+        # item is guaranteed PENDING too, so a plain index/status check is
+        # enough here - see module docstring.
+        can_edit_selected = (
+            selected_item is not None and selected_item.status == ScheduleItemStatus.PENDING
+        )
+
+        self.run_btn.setEnabled(not is_running)
+        # Adding is always safe: new items are clamped into the PENDING
+        # suffix regardless of what's selected (see add_item()).
+        self.add_btn.setEnabled(True)
+        self.edit_btn.setEnabled(can_edit_selected)
+        self.delete_btn.setEnabled(can_edit_selected)
+        self.move_up_btn.setEnabled(can_edit_selected and idx not in (None, 0))
+        self.move_down_btn.setEnabled(
+            can_edit_selected and idx is not None and idx < len(items) - 1
+        )
+
+        if self._guard.enabled:
+            state = "OK" if self._guard.is_clear() else "PAUSED - waiting to resume"
+            value = self._guard.current_value
+            value_text = f"{value:g}" if value is not None else "?"
+            self.guard_status_label.setText(
+                f"Auto-pause: {self._guard.device_name} = {value_text}  "
+                f"(pause < {self._guard.pause_below:g}, resume > {self._guard.resume_above:g})  [{state}]"
+            )
+            self.guard_status_label.show()
+        else:
+            self.guard_status_label.hide()
+
+    def _on_selection_changed(self, row: int):
+        item = self.list_widget.item(row) if row >= 0 else None
+        self._selected_item_id = item.data(_ITEM_ID_ROLE) if item is not None else None
+        self._refresh_ui()
+
+    # ---- notifications / auto-pause settings ---- #
+    @SafeSlot()
+    def _on_notifications_clicked(self):
+        dialog = NotificationSettingsDialog(self.schedule.notifications, parent=self)
+        if dialog.exec_() != QDialog.Accepted:
+            return
+        with self._lock:
+            self.schedule.notifications = dialog.result_settings()
+            self._persist_locked()
+        self._refresh_ui()
+
+    @SafeSlot()
+    def _on_guard_clicked(self):
+        device_names = sorted(self.dev.keys())
+        dialog = GuardSettingsDialog(self.schedule.guard, device_names, parent=self)
+        if dialog.exec_() != QDialog.Accepted:
+            return
+        with self._lock:
+            self.schedule.guard = dialog.result_settings()
+            self._persist_locked()
+        self._guard.configure(self.schedule.guard)
+        self._refresh_ui()
+
+    def _apply_guard_settings(self):
+        """Re-subscribe the guard after `self.schedule.guard` changes (e.g. from a remote update)."""
+        self._guard.configure(self.schedule.guard)
+
+    @SafeSlot(float)
+    def _on_guard_paused(self, value: float):
+        logger.info(
+            "Auto-pause guard tripped for schedule '%s' (value=%s); interrupting the running scan, if any.",
+            self.schedule_name,
+            value,
+        )
+        if self._current_report is not None and self._current_item_kind == "scan":
+            self._guard_interrupt_requested = True
+            try:
+                self._current_report.cancel()
+            except Exception:  # pylint: disable=broad-except
+                logger.exception("Failed to cancel the running scan for the auto-pause guard")
+        self.schedule_changed.emit()
+
+    @SafeSlot(float)
+    def _on_guard_resumed(self, value: float):
+        logger.info(
+            "Auto-pause guard cleared for schedule '%s' (value=%s); the execution loop will "
+            "resume the next scan item automatically.",
+            self.schedule_name,
+            value,
+        )
+        self.schedule_changed.emit()
+
+    # ---- UI-triggered edit actions ---- #
+    @SafeSlot()
+    def _on_add_clicked(self):
+        dialog = ScheduleItemDialog(self.scans, self.dev, parent=self, client=self.client)
+        if dialog.exec_() != QDialog.Accepted:
+            return
+        result = dialog.result()
+        with self._lock:
+            idx = self._index_of_locked(self._selected_item_id)
+        insert_at = None if idx is None else idx + 1
+        new_id = self.add_item(
+            result["command"], index=insert_at, kind=result["kind"], form_state=result["form_state"]
+        )
+        self._selected_item_id = new_id
+        self._refresh_ui()
+
+    @SafeSlot()
+    def _on_edit_clicked(self):
+        if self._selected_item_id is None:
+            return
+        with self._lock:
+            idx = self._index_of_locked(self._selected_item_id)
+            item = self.schedule.items[idx] if idx is not None else None
+            initial = (
+                {"kind": item.kind, "command": item.command, "form_state": item.form_state}
+                if item is not None
+                else None
+            )
+        if initial is None:
+            return
+
+        dialog = ScheduleItemDialog(
+            self.scans, self.dev, parent=self, initial=initial, client=self.client
+        )
+        if dialog.exec_() != QDialog.Accepted:
+            return
+        result = dialog.result()
+        try:
+            self.edit_item(
+                self._selected_item_id,
+                result["command"],
+                kind=result["kind"],
+                form_state=result["form_state"],
+            )
+        except RuntimeError as exc:
+            QMessageBox.warning(self, "Cannot edit item", str(exc))
+            return
+        self._refresh_ui()
+
+    @SafeSlot()
+    def _on_delete_clicked(self):
+        if self._selected_item_id is None:
+            return
+        confirm = QMessageBox.question(
+            self, "Delete schedule item", "Remove the selected item from the schedule?"
+        )
+        if confirm != QMessageBox.Yes:
+            return
+        try:
+            self.delete_item(self._selected_item_id)
+        except RuntimeError as exc:
+            QMessageBox.warning(self, "Cannot delete item", str(exc))
+            return
+        self._selected_item_id = None
+        self._refresh_ui()
+
+    @SafeSlot()
+    def _on_move_up_clicked(self):
+        if self._selected_item_id is None:
+            return
+        try:
+            self.move_item_up(self._selected_item_id)
+        except RuntimeError as exc:
+            QMessageBox.warning(self, "Cannot move item", str(exc))
+
+    @SafeSlot()
+    def _on_move_down_clicked(self):
+        if self._selected_item_id is None:
+            return
+        try:
+            self.move_item_down(self._selected_item_id)
+        except RuntimeError as exc:
+            QMessageBox.warning(self, "Cannot move item", str(exc))
+
+    # ------------------------------------------------------------------ #
+    # RPC-exposed actions (USER_ACCESS)
+    # ------------------------------------------------------------------ #
+    @SafeSlot()
+    def run_schedule(self):
+        """Run the schedule, continuing from wherever it last left off."""
+        with self._lock:
+            if self.schedule.is_running:
+                return
+            self._abort_requested = False
+            self.schedule.is_running = True
+            self._persist_locked()
+        self.schedule_changed.emit()
+        self.submit_task(
+            self._run_all, on_complete=self._on_run_finished, on_failed=self._on_run_failed
+        )
+
+    @SafeSlot()
+    def abort_schedule(self):
+        """Request that the schedule stop after the current item."""
+        self._abort_requested = True
+        if self._current_report is not None:
+            try:
+                self._current_report.cancel()
+            except Exception:  # pylint: disable=broad-except
+                logger.exception("Failed to cancel the currently running schedule item")
+
+    @SafeSlot()
+    def reset_schedule(self):
+        """Clear all execution state and start the schedule over from item 1."""
+        with self._lock:
+            if self.schedule.is_running:
+                logger.info("Cannot reset schedule, as schedule is running!")
+                return
+            for item in self.schedule.items:
+                logger.info(f"Reset schedule item {item}")
+                item.status = ScheduleItemStatus.PENDING
+                item.request_id = None
+                item.scan_id = None
+                item.scan_number = None
+                item.error = None
+                item.started_at = None
+                item.finished_at = None
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+    def get_status(self) -> dict:
+        """RPC-exposed: current schedule state, e.g. for another widget or a script."""
+        with self._lock:
+            return self.schedule.model_dump(mode="json")
+
+    # ------------------------------------------------------------------ #
+    # RPC-exposed schedule editing (add / edit / move / delete)
+    #
+    # All of these only ever touch the PENDING suffix of the schedule -
+    # see the "protected prefix" section of the module docstring. They can
+    # be called at any time, including while the schedule is running.
+    # ------------------------------------------------------------------ #
+    def add_item(
+        self,
+        command: str,
+        index: int | None = None,
+        kind: str = "custom",
+        form_state: dict | None = None,
+    ) -> str:
+        """
+        RPC-exposed: insert a new, PENDING command into the schedule.
+        Safe to call while the schedule is running.
+
+        Args:
+            command: command text, evaluated the same way as the existing
+                items (against `scans`/`dev`) once the schedule runs.
+            index: position to insert at (0 = first). Clamped so the item
+                can never land before something already running or
+                finished. Defaults to appending at the end.
+            kind/form_state: optional structured description of how
+                `command` was built (see `schedule_item.ScheduleItem`),
+                used to reopen the Edit dialog pre-filled. Leave as
+                defaults for a plain, hand-typed command.
+
+        Returns:
+            The new item's item_id.
+        """
+        command = command.strip()
+        if not command:
+            raise ValueError("command must not be empty")
+
+        with self._lock:
+            protected = self._protected_prefix_length_locked()
+            item = ScheduleItem(
+                item_id=str(uuid.uuid4()), command=command, kind=kind, form_state=form_state
+            )
+            if index is None or index >= len(self.schedule.items):
+                self.schedule.items.append(item)
+            else:
+                self.schedule.items.insert(max(index, protected), item)
+            self._persist_locked()
+        self.schedule_changed.emit()
+        return item.item_id
+
+    def edit_item(
+        self, item_id: str, command: str, kind: str = "custom", form_state: dict | None = None
+    ):
+        """
+        RPC-exposed: change the command of an item that has not started
+        yet. Raises `RuntimeError` for an item that is already
+        running/finished.
+        """
+        command = command.strip()
+        if not command:
+            raise ValueError("command must not be empty")
+
+        with self._lock:
+            idx = self._index_of_locked(item_id)
+            if idx is None:
+                return
+            item = self.schedule.items[idx]
+            if item.status != ScheduleItemStatus.PENDING:
+                raise RuntimeError("Only items that have not started yet (PENDING) can be edited.")
+            item.command = command
+            item.kind = kind
+            item.form_state = form_state
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+    def delete_item(self, item_id: str):
+        """
+        RPC-exposed: remove an item that has not started yet. Raises
+        `RuntimeError` for an item that is already running/finished.
+        """
+        with self._lock:
+            idx = self._index_of_locked(item_id)
+            if idx is None:
+                return
+            if self.schedule.items[idx].status != ScheduleItemStatus.PENDING:
+                raise RuntimeError("Only items that have not started yet (PENDING) can be deleted.")
+            del self.schedule.items[idx]
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+    def move_item(self, item_id: str, new_index: int):
+        """
+        RPC-exposed: move an item that has not started yet to a new
+        position (0 = first, but never before something already
+        running/finished). Raises `RuntimeError` for an item that is
+        already running/finished.
+        """
+        with self._lock:
+            idx = self._index_of_locked(item_id)
+            if idx is None:
+                return
+            if self.schedule.items[idx].status != ScheduleItemStatus.PENDING:
+                raise RuntimeError("Only items that have not started yet (PENDING) can be moved.")
+            protected = self._protected_prefix_length_locked()
+            item = self.schedule.items.pop(idx)
+            new_index = max(protected, min(new_index, len(self.schedule.items)))
+            self.schedule.items.insert(new_index, item)
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+    def move_item_up(self, item_id: str):
+        """RPC-exposed: swap an item with the one directly before it."""
+        with self._lock:
+            idx = self._index_of_locked(item_id)
+            target = idx - 1 if idx is not None and idx > 0 else None
+        if target is not None:
+            self.move_item(item_id, target)
+
+    def move_item_down(self, item_id: str):
+        """RPC-exposed: swap an item with the one directly after it."""
+        with self._lock:
+            idx = self._index_of_locked(item_id)
+            n = len(self.schedule.items)
+            target = idx + 1 if idx is not None and idx < n - 1 else None
+        if target is not None:
+            self.move_item(item_id, target)
+
+    # ------------------------------------------------------------------ #
+    # execution - runs in a background task (BECConnector.submit_task,
+    # backed by a QThreadPool worker thread; UI updates below go through
+    # the `schedule_changed` Qt signal so they are marshalled back onto
+    # the GUI thread instead of touching widgets directly)
+    # ------------------------------------------------------------------ #
+    def _pick_next_runnable_locked(self):
+        """Must be called while holding `self._lock`; see `schedule_logic.pick_next_runnable`."""
+        return pick_next_runnable(self.schedule.items)
+
+    def _run_all(self):
+        namespace = {"scans": self.scans, "dev": self.dev}
+        while not self._abort_requested:
+            with self._lock:
+                next_item = self._pick_next_runnable_locked()
+            if next_item is None or next_item == "stop":
+                break
+            if next_item.status == ScheduleItemStatus.RUNNING:
+                self._await_running_item(next_item)
+            else:
+                self._execute_item(next_item, namespace)
+
+        with self._lock:
+            self.schedule.is_running = False
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+    def _execute_item(self, item: ScheduleItem, namespace: dict):
+        if item.kind == "scan":
+            # Block here (item stays PENDING - nothing "started" yet) until
+            # the guard is clear or the operator aborts. See "Guard
+            # interaction" in the module docstring.
+            cleared = self._guard.wait_until_clear(should_abort=lambda: self._abort_requested)
+            self.schedule_changed.emit()
+            if not cleared:
+                return
+
+        with self._lock:
+            item.status = ScheduleItemStatus.RUNNING
+            item.error = None
+            item.started_at = time.time()
+            self._persist_locked()
+        self.schedule_changed.emit()
+
+        try:
+            report = eval(
+                item.command, {"__builtins__": {}}, namespace
+            )  # noqa: S307  pylint: disable=eval-used
+            self._current_report = report
+            self._current_item_kind = item.kind
+
+            request = getattr(report, "request", None)
+            with self._lock:
+                item.request_id = getattr(request, "requestID", None)
+                # persist the request_id right away, before the
+                # (potentially long) wait below, so a reconnect can find
+                # it even if this process disappears mid-scan.
+                self._persist_locked()
+
+            report.wait()
+
+            scan = getattr(report, "scan", None)
+            with self._lock:
+                item.scan_id = getattr(scan, "scan_id", None) if scan else None
+                item.scan_number = getattr(scan, "scan_number", None) if scan else None
+                item.status = (
+                    ScheduleItemStatus.ABORTED
+                    if self._abort_requested
+                    else ScheduleItemStatus.COMPLETED
+                )
+        except Exception:  # pylint: disable=broad-except
+            with self._lock:
+                if self._guard_interrupt_requested:
+                    # Interrupted by the auto-pause guard, not a real
+                    # failure/operator abort - reset to PENDING so the
+                    # normal execution loop retries it once the guard
+                    # clears, instead of leaving it in a terminal state.
+                    item.status = ScheduleItemStatus.PENDING
+                    item.request_id = None
+                    item.scan_id = None
+                    item.scan_number = None
+                    item.error = None
+                    item.started_at = None
+                    self._guard_interrupt_requested = False
+                elif self._abort_requested:
+                    item.status = ScheduleItemStatus.ABORTED
+                else:
+                    item.status = ScheduleItemStatus.FAILED
+                    item.error = traceback.format_exc()
+                    logger.error(f"Schedule item failed: {item.command}\n{item.error}")
+        finally:
+            with self._lock:
+                if item.status != ScheduleItemStatus.PENDING:
+                    item.finished_at = time.time()
+                self._persist_locked()
+                final_status = item.status
+            self._current_report = None
+            self._current_item_kind = None
+            self.schedule_changed.emit()
+
+        if final_status in (ScheduleItemStatus.COMPLETED, ScheduleItemStatus.FAILED):
+            notify_item_finished(
+                self.connector,
+                self.schedule_name,
+                item.kind,
+                item.command,
+                success=(final_status == ScheduleItemStatus.COMPLETED),
+                settings=self.schedule.notifications,
+                error=item.error,
+            )
+
+    def _await_running_item(self, item: ScheduleItem):
+        queue_storage = self.client.queue.queue_storage
+        while not self._abort_requested:
+            queue_item = queue_storage.find_queue_item_by_requestID(item.request_id)
+            if queue_item is None or not self._is_queue_item_active(queue_item):
+                with self._lock:
+                    item.status = ScheduleItemStatus.COMPLETED
+                    item.finished_at = time.time()
+                    self._persist_locked()
+                self.schedule_changed.emit()
+                return
+            time.sleep(0.5)
+
+    @SafeSlot()
+    def _on_run_finished(self):
+        self._refresh_ui()
+
+    @SafeSlot(str)
+    def _on_run_failed(self, error: str):
+        logger.error(f"Schedule execution task failed unexpectedly: {error}")
+        with self._lock:
+            self.schedule.is_running = False
+            self._persist_locked()
+        self._refresh_ui()
+
+    # ------------------------------------------------------------------ #
+    # cleanup
+    # ------------------------------------------------------------------ #
+    def cleanup(self):
+        # Note: we deliberately do NOT abort a running schedule here. The
+        # whole point of the Redis-backed design is that closing this
+        # widget must not interrupt anything already submitted to the
+        # scan/device server - see the module docstring.
+        self._guard.cleanup()
+        super().cleanup()  # also disconnects all bec_dispatcher subscriptions for us

debye_bec/bec_widgets/widgets/scheduler/scheduler.pyproject

@@ -0,0 +1 @@
+{'files': ['scheduler.py']}

debye_bec/bec_widgets/widgets/scheduler/scheduler_plugin.py

@@ -0,0 +1,57 @@
+# Copyright (C) 2022 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
+
+from bec_widgets.utils.bec_designer import designer_material_icon
+from qtpy.QtDesigner import QDesignerCustomWidgetInterface
+from qtpy.QtWidgets import QWidget
+
+from .scheduler import Scheduler
+
+DOM_XML = """
+<ui language='c++'>
+    <widget class='Scheduler' name='scheduler'>
+    </widget>
+</ui>
+"""
+
+
+class SchedulerPlugin(QDesignerCustomWidgetInterface):  # pragma: no cover
+    def __init__(self):
+        super().__init__()
+        self._form_editor = None
+
+    def createWidget(self, parent):
+        if parent is None:
+            return QWidget()
+        t = Scheduler(parent)
+        return t
+
+    def domXml(self):
+        return DOM_XML
+
+    def group(self):
+        return ""
+
+    def icon(self):
+        return designer_material_icon(Scheduler.ICON_NAME)
+
+    def includeFile(self):
+        return "scheduler"
+
+    def initialize(self, form_editor):
+        self._form_editor = form_editor
+
+    def isContainer(self):
+        return False
+
+    def isInitialized(self):
+        return self._form_editor is not None
+
+    def name(self):
+        return "Scheduler"
+
+    def toolTip(self):
+        return "Scheduler"
+
+    def whatsThis(self):
+        return self.toolTip()