feat: add scan scheduler editor

bec_widgets/applications/main_app.py

@@ -24,6 +24,7 @@ from bec_widgets.utils.screen_utils import (
 )
 from bec_widgets.widgets.containers.dock_area.profile_utils import is_experimental_features_enabled
 from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
+from bec_widgets.widgets.control.scan_scheduler import ScanScheduler
 
 startup_profiler.mark("module imports")
 
@@ -94,6 +95,14 @@ class BECMainApp(BECMainWindow):
             widget=self.device_manager,
             mini_text="DM",
         )
+        self.scan_scheduler = ScanScheduler(parent=self)
+        self.add_view(
+            icon="account_tree",
+            title="Scan Scheduler",
+            widget=self.scan_scheduler,
+            mini_text="Scans",
+            view_id="scan_scheduler",
+        )
         if is_experimental_features_enabled():
             self.developer_view = DeveloperView(self)
             self.add_view(

bec_widgets/cli/client.py

@@ -54,6 +54,7 @@ _Widgets = {
     "SBBMonitor": "SBBMonitor",
     "ScanControl": "ScanControl",
     "ScanProgressBar": "ScanProgressBar",
+    "ScanScheduler": "ScanScheduler",
     "ScatterWaveform": "ScatterWaveform",
     "SignalLabel": "SignalLabel",
     "TextBox": "TextBox",
@@ -5426,6 +5427,43 @@ class ScanProgressBar(RPCBase):
         """
 
 
+class ScanScheduler(RPCBase):
+    """Edit nested components and execute validated sequences through procedures."""
+
+    _IMPORT_MODULE = "bec_widgets.widgets.control.scan_scheduler.scan_scheduler"
+
+    @rpc_call
+    def load_json(self, document: "str", definitions: "str" = "[]") -> "None":
+        """
+        Load a structurally valid document and reusable definitions into the editor.
+
+        Args:
+            document (str): Sequence document JSON.
+            definitions (str): JSON list of reusable definitions. Defaults to an empty library.
+        """
+
+    @rpc_call
+    def document_json(self) -> "str":
+        """
+        Return the current edited document with structural schema validation.
+
+        Returns:
+            str: Authoritative sequence JSON for saving or submission.
+        """
+
+    @rpc_call
+    def validate_sequence(self) -> "None":
+        """
+        Validate the entire draft and refresh all exports on a background task.
+        """
+
+    @rpc_call
+    def execute_sequence(self) -> "None":
+        """
+        Validate and execute the current sequence through the existing procedure API.
+        """
+
+
 class ScatterCurve(RPCBase):
     """Scatter curve item for the scatter waveform widget."""
 

bec_widgets/cli/designer_plugins.py

@@ -100,6 +100,7 @@ designer_plugins = {
         "bec_widgets.widgets.progress.scan_progressbar.scan_progressbar",
         "ScanProgressBar",
     ),
+    "ScanScheduler": ("bec_widgets.widgets.control.scan_scheduler.scan_scheduler", "ScanScheduler"),
     "ScatterWaveform": (
         "bec_widgets.widgets.plots.scatter_waveform.scatter_waveform",
         "ScatterWaveform",
@@ -163,6 +164,7 @@ widget_icons = {
     "ScanIndexComboBox": "history",
     "ScanMetadata": "list_alt",
     "ScanProgressBar": "timelapse",
+    "ScanScheduler": "account_tree",
     "ScatterWaveform": "scatter_plot",
     "SignalComboBox": "list_alt",
     "SignalLabel": "scoreboard",

bec_widgets/examples/scan_scheduler.py

@@ -0,0 +1,58 @@
+"""Open a scheduler with an editable example using a running BEC deployment.
+
+Run with ``python -m bec_widgets.examples.scan_scheduler``. Opening the example does
+not execute it; inspect the description before choosing Execute.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from bec_lib.scan_sequencer.components import ForEach, Repeat, ScanCall, Schedule, Sequence, Set
+from qtpy.QtWidgets import QApplication
+
+from bec_widgets.widgets.control.scan_scheduler import ScanScheduler
+
+
+def example_sequence() -> Schedule:
+    """Compose four time scans interleaved with alternating device moves.
+
+    Returns:
+        Schedule: Editable example requiring samx and the advertised time_scan scan.
+    """
+    return Schedule(
+        id="alternating_positions",
+        revision=1,
+        root=Repeat(
+            id="repeat",
+            count=2,
+            body=ForEach(
+                id="positions",
+                variable="position",
+                values=[1, -1],
+                body=Sequence(
+                    id="steps",
+                    children=[
+                        Set(id="move", device="samx", value={"ref": "position"}, wait=True),
+                        ScanCall(id="scan", name="time_scan", positional=[1, 0.1]),
+                    ],
+                ),
+            ),
+        ),
+    )
+
+
+def main() -> None:
+    """Open the scheduler; execution requires an explicit Execute action."""
+    app = QApplication.instance() or QApplication(sys.argv)
+    widget = ScanScheduler()
+    widget.setWindowTitle("BEC Scan Scheduler")
+    widget.resize(1400, 1000)
+    widget.load_json(example_sequence().model_dump_json())
+    widget.show()
+    app.exec()
+    widget.cleanup()
+
+
+if __name__ == "__main__":
+    main()

bec_widgets/widgets/control/scan_control/scan_control.py

@@ -80,8 +80,23 @@ class ScanControl(BECWidget, QWidget):
         gui_id: str | None = None,
         allowed_scans: list | None = None,
         default_scan: str | None = None,
+        scan_info: dict | None = None,
         **kwargs,
     ):
+        """Initialize scan controls from subscribed metadata or the available-scans endpoint.
+
+        Args:
+            parent (QWidget | None): Owning widget.
+            client (BECClient | None): Existing BEC client.
+            config (ScanControlConfig | dict | None): Widget configuration.
+            gui_id (str | None): RPC widget identity.
+            allowed_scans (list | None): Optional scan-selection filter.
+            default_scan (str | None): Initial selected scan.
+            scan_info (dict | None): Already subscribed scan metadata, avoiding a Redis read
+                during embedded editor construction. Defaults to reading the existing endpoint.
+            **kwargs (object): Additional BECWidget arguments.
+        """
+        self._available_scans_override = scan_info
         if config is None:
             config = ScanControlConfig(
                 widget_class=self.__class__.__name__, allowed_scans=allowed_scans
@@ -254,9 +269,11 @@ class ScanControl(BECWidget, QWidget):
 
     def populate_scans(self):
         """Populates the scan selection combo box with available scans from BEC session."""
-        self.available_scans = self.client.connector.get(
-            MessageEndpoints.available_scans()
-        ).resource
+        self.available_scans = (
+            self._available_scans_override
+            if self._available_scans_override is not None
+            else self.client.connector.get(MessageEndpoints.available_scans()).resource
+        )
         self._update_scan_selector()
 
     def _supported_scan_names(self) -> list[str]:
@@ -768,7 +785,7 @@ class ScanControl(BECWidget, QWidget):
             hide(bool): Hide or show the metadata form.
         """
         self._hide_metadata = hide
-        self._metadata_form.setVisible(not hide)
+        self._metadata_group.setVisible(not hide)
 
     @SafeProperty(bool)
     def hide_optional_metadata(self):

bec_widgets/widgets/control/scan_scheduler/__init__.py

@@ -0,0 +1,5 @@
+"""Sequence composition and procedure submission widget."""
+
+from .scan_scheduler import ScanScheduler
+
+__all__ = ["ScanScheduler"]

bec_widgets/widgets/control/scan_scheduler/document.py

@@ -0,0 +1,190 @@
+"""Editable sequence trees owned by the scheduler interface."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from uuid import uuid4
+
+from bec_lib.scan_sequencer.components import Schedule
+
+
+def component(kind: str) -> dict:
+    """Create an editable component with a unique identity.
+
+    Args:
+        kind (str): Backend component discriminator.
+
+    Returns:
+        dict: Component template. Scan and device targets require user selection.
+    """
+    node = {"id": f"{kind}_{uuid4().hex[:12]}", "type": kind}
+    if kind == "sequence":
+        node["children"] = []
+    elif kind in {"repeat", "for_each", "while"}:
+        node["body"] = component("sequence")
+        node.update(
+            {
+                "repeat": {"count": 1},
+                "for_each": {"variable": "value", "values": []},
+                "while": {"condition": False},
+            }[kind]
+        )
+    elif kind == "set":
+        node.update(device="", value=0, wait=True, bind_status=None)
+    elif kind == "scan_call":
+        node.update(name="", positional=[], arguments={})
+    elif kind == "component_call":
+        node.update(definition={"id": "", "version": 1}, arguments={})
+    else:
+        raise ValueError(f"Unknown component: {kind}")
+    return node
+
+
+class SequenceDraft:
+    """Hold an editable tree independently of submitted procedure payloads.
+
+    Args:
+        document (dict | None): Initial sequence document or a new empty sequence.
+    """
+
+    def __init__(self, document: dict | None = None) -> None:
+        self.document = (
+            deepcopy(document)
+            if document is not None
+            else {
+                "schema_version": 1,
+                "type": "sequence",
+                "id": "new_sequence",
+                "revision": 1,
+                "root": component("sequence"),
+            }
+        )
+
+    def node(self, path: tuple) -> dict:
+        """Look up a component by its structural path.
+
+        Args:
+            path (tuple): Keys and child indices relative to the root.
+
+        Returns:
+            dict: Mutable component in this draft.
+        """
+        value = self.document["root"]
+        for key in path:
+            value = value[key]
+        return value
+
+    def replace(self, path: tuple, node: dict) -> None:
+        """Replace one component.
+
+        Args:
+            path (tuple): Component path.
+            node (dict): Replacement component.
+        """
+        if not path:
+            self.document["root"] = node
+        else:
+            parent = self.node(path[:-1])
+            parent[path[-1]] = node
+
+    def add(self, path: tuple, kind: str) -> tuple:
+        """Append inside a sequence or loop; insert after a selected sequence child.
+
+        Args:
+            path (tuple): Selected component path.
+            kind (str): Component type to insert.
+
+        Returns:
+            tuple: Inserted component's path.
+        """
+        node = self.node(path)
+        if "body" in node:
+            if node["body"]["type"] != "sequence":
+                wrapper = component("sequence")
+                wrapper["children"] = [node["body"]]
+                node["body"] = wrapper
+            path += ("body",)
+            node = node["body"]
+        if node["type"] == "sequence":
+            index = len(node["children"])
+            node["children"].append(component(kind))
+            return path + ("children", index)
+        if len(path) >= 2 and path[-2] == "children":
+            index = path[-1] + 1
+            self.node(path[:-2])["children"].insert(index, component(kind))
+            return path[:-1] + (index,)
+        raise ValueError("Select a sequence, or wrap this step in a sequence first.")
+
+    def wrap(self, path: tuple, kind: str) -> None:
+        """Wrap the selected component in a sequence or loop.
+
+        Args:
+            path (tuple): Selected component path.
+            kind (str): Sequence, repeat, for_each or while.
+        """
+        if kind not in {"sequence", "repeat", "for_each", "while"}:
+            raise ValueError("Only a sequence or loop can wrap a step.")
+        wrapper = component(kind)
+        if kind == "sequence":
+            wrapper["children"] = [self.node(path)]
+        else:
+            wrapper["body"] = self.node(path)
+        self.replace(path, wrapper)
+
+    def remove(self, path: tuple) -> None:
+        """Remove a child, replacing a required root or body with an empty sequence.
+
+        Args:
+            path (tuple): Selected component path.
+        """
+        if len(path) >= 2 and path[-2] == "children":
+            self.node(path[:-2])["children"].pop(path[-1])
+        else:
+            self.replace(path, component("sequence"))
+
+    def move(self, path: tuple, offset: int) -> tuple:
+        """Move a component amongst its sequence siblings.
+
+        Args:
+            path (tuple): Selected child path.
+            offset (int): Relative sibling position.
+
+        Returns:
+            tuple: New component path.
+        """
+        if len(path) < 2 or path[-2] != "children":
+            return path
+        children = self.node(path[:-2])["children"]
+        target = path[-1] + offset
+        if 0 <= target < len(children):
+            children.insert(target, children.pop(path[-1]))
+            return path[:-1] + (target,)
+        return path
+
+    def validated(self) -> Schedule:
+        """Validate the document's structural schema.
+
+        Returns:
+            Schedule: Typed document; semantic validation is performed by the sequencer.
+        """
+        return Schedule.model_validate(self.document)
+
+    def variables(self, path: tuple) -> list[str]:
+        """Return visible loop variables, with inner bindings shadowing outer ones.
+
+        Args:
+            path (tuple): Selected component path.
+
+        Returns:
+            list[str]: Names visible inside enclosing for_each bodies.
+        """
+        names = []
+        for length, part in enumerate(path):
+            if part == "body":
+                ancestor = self.node(path[:length])
+                if ancestor["type"] == "for_each":
+                    name = ancestor["variable"]
+                    if name in names:
+                        names.remove(name)
+                    names.append(name)
+        return names

bec_widgets/widgets/control/scan_scheduler/inspector.py

@@ -0,0 +1,386 @@
+"""Component field editors with explicit literal and scoped-variable inputs."""
+
+from __future__ import annotations
+
+import json
+
+from qtpy.QtCore import Qt, Signal
+from qtpy.QtWidgets import (
+    QCheckBox,
+    QComboBox,
+    QFormLayout,
+    QHBoxLayout,
+    QLabel,
+    QLineEdit,
+    QPlainTextEdit,
+    QPushButton,
+    QSpinBox,
+    QVBoxLayout,
+    QWidget,
+)
+
+from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter
+
+from .presentation import LABELS, Section
+from .scan_editor import ScanDefinitionEditor
+
+# Qt signal descriptors are not understood by pylint.
+# pylint: disable=no-member
+
+
+class ValueInput(QWidget):
+    """Edit a JSON value or an explicit loop-variable reference.
+
+    Args:
+        value (object): Initial JSON value.
+        variables (list[str]): Visible loop variable names.
+        parent (QWidget | None): Owning widget.
+    """
+
+    changed = Signal()
+
+    def __init__(self, value: object, variables: list[str], parent=None) -> None:
+        super().__init__(parent)
+        layout = QHBoxLayout(self)
+        layout.setContentsMargins(0, 0, 0, 0)
+        self.mode = QComboBox(self)
+        self.mode.addItem("Value", None)
+        for name in variables:
+            self.mode.addItem(f"Variable: {name}", name)
+        self.allow_encoding = True
+        self.encoding = QComboBox(self)
+        for label, kind in (
+            ("Text", "text"),
+            ("Number", "number"),
+            ("Boolean", "boolean"),
+            ("JSON", "json"),
+        ):
+            self.encoding.addItem(label, kind)
+        kind = (
+            "text"
+            if isinstance(value, str)
+            else (
+                "boolean"
+                if isinstance(value, bool)
+                else "number" if isinstance(value, (int, float)) else "json"
+            )
+        )
+        self.encoding.setCurrentIndex(self.encoding.findData(kind))
+        self.value_edit = QLineEdit(
+            value if kind == "text" else json.dumps(value, ensure_ascii=False, allow_nan=False),
+            self,
+        )
+        self.value_edit.setPlaceholderText('e.g. 1.0, "Circ+", [1, 2], {"type": "status_field", …}')
+        layout.addWidget(self.mode)
+        layout.addWidget(self.value_edit, 1)
+        layout.addWidget(self.encoding)
+        self.mode.currentIndexChanged.connect(self._mode_changed)
+        self.mode.currentIndexChanged.connect(lambda *_: self.changed.emit())
+        self.encoding.currentIndexChanged.connect(lambda *_: self.changed.emit())
+        self.value_edit.editingFinished.connect(self.changed.emit)
+        if isinstance(value, dict) and set(value) == {"ref"}:
+            index = self.mode.findData(value["ref"])
+            if index >= 0:
+                self.mode.setCurrentIndex(index)
+
+    def _mode_changed(self) -> None:
+        self.value_edit.setVisible(self.mode.currentData() is None)
+        self.encoding.setVisible(self.mode.currentData() is None and self.allow_encoding)
+
+    def value(self) -> object:
+        """Read the explicit reference or parsed literal/expression.
+
+        Returns:
+            object: JSON value for the backend.
+        """
+        name = self.mode.currentData()
+        if name is not None:
+            return {"ref": name}
+        text = self.value_edit.text()
+        kind = self.encoding.currentData()
+        if kind == "text":
+            return text
+        value = json.loads(text)
+        if kind == "number" and (isinstance(value, bool) or not isinstance(value, (int, float))):
+            raise ValueError("Enter a number, or change the input type.")
+        if kind == "boolean" and not isinstance(value, bool):
+            raise ValueError("Enter true or false.")
+        return value
+
+
+class ValuesInput(QWidget):
+    """Edit ordered loop values as individual typed inputs.
+
+    Args:
+        values (list): Initial loop values.
+        variables (list[str]): Visible outer loop bindings.
+        parent (QWidget | None): Owning widget.
+    """
+
+    changed = Signal()
+
+    def __init__(self, values: list, variables: list[str], parent=None) -> None:
+        super().__init__(parent)
+        self.variables = variables
+        self.rows = []
+        self.rows_layout = QVBoxLayout(self)
+        self.rows_layout.setContentsMargins(0, 0, 0, 0)
+        self.add_button = QPushButton("+ Add value", self)
+        self.add_button.clicked.connect(lambda: self._add(""))
+        self.rows_layout.addWidget(self.add_button)
+        for value in values:
+            self._add(value)
+
+    def _add(self, value):
+        row = QWidget(self)
+        layout = QHBoxLayout(row)
+        layout.setContentsMargins(0, 0, 0, 0)
+        editor = ValueInput(value, self.variables, row)
+        remove = QPushButton("×", row)
+        remove.setFixedWidth(28)
+        layout.addWidget(editor, 1)
+        layout.addWidget(remove)
+        self.rows_layout.insertWidget(len(self.rows), row)
+        self.rows.append((row, editor))
+        remove.clicked.connect(lambda: self._remove(row, editor))
+        editor.changed.connect(self.changed.emit)
+        self.changed.emit()
+
+    def _remove(self, row, editor):
+        self.rows.remove((row, editor))
+        self.rows_layout.removeWidget(row)
+        row.deleteLater()
+        self.changed.emit()
+
+    def value(self) -> list:
+        """Read loop values in displayed order.
+
+        Returns:
+            list: JSON loop values.
+        """
+        return [editor.value() for _, editor in self.rows]
+
+
+class ComponentInspector(QWidget):
+    """Show the backend's component documentation and editable fields.
+
+    Args:
+        parent (QWidget | None): Owning widget.
+        client (BECClient | None): Shared BEC client for the embedded ScanControl.
+    """
+
+    changed = Signal()
+
+    def __init__(self, parent=None, client=None) -> None:
+        super().__init__(parent)
+        self.client = client
+        layout = QVBoxLayout(self)
+        layout.setAlignment(Qt.AlignTop)
+        self.breadcrumb = QLabel(self)
+        self.help = QLabel(self)
+        self.help.setWordWrap(True)
+        self.scope = QLabel(self)
+        self.scope.setWordWrap(True)
+        self.breadcrumb.setObjectName("breadcrumb")
+        self.help.setObjectName("stepHeading")
+        self.scope.setObjectName("scopeChip")
+        layout.addWidget(self.breadcrumb)
+        layout.addWidget(self.help)
+        layout.addWidget(self.scope)
+        for label in (self.help, self.scope, self.breadcrumb):
+            label.setTextFormat(Qt.PlainText)
+        self.form = QFormLayout()
+        self.form.setVerticalSpacing(14)
+        self.form.setFieldGrowthPolicy(QFormLayout.AllNonFixedFieldsGrow)
+        layout.addLayout(self.form)
+        self.metadata = QPlainTextEdit(self)
+        self.metadata.setReadOnly(True)
+        self.metadata.setMaximumHeight(200)
+        self.metadata.setPlaceholderText("Advertised scan signature and documentation")
+        self.documentation = Section("Component documentation", self.metadata, self)
+        layout.addWidget(self.documentation)
+        layout.addStretch()
+        self.fields = {}
+        self.node = {}
+        self.scan_arguments = None
+        self.raw_arguments = None
+
+    # The inspector receives independently owned editor metadata.
+    # pylint: disable=too-many-arguments,too-many-positional-arguments
+    def display(
+        self,
+        node: dict,
+        schema: dict,
+        variables: list[str],
+        scans: dict,
+        devices: list[str],
+        breadcrumb: str,
+    ) -> None:
+        """Populate fields using component schema and subscribed scan metadata.
+
+        Args:
+            node (dict): Selected component.
+            schema (dict): Component JSON schema.
+            variables (list[str]): Visible loop bindings.
+            scans (dict): Advertised scan metadata.
+            devices (list[str]): Available device names; dotted signal names remain editable.
+            breadcrumb (str): Selected component's nesting path.
+        """
+        self.cleanup()
+        self.node = node
+        self.scan_arguments = None
+        self.raw_arguments = None
+        self.fields = {}
+        while self.form.rowCount():
+            self.form.removeRow(0)
+        self.breadcrumb.setText(breadcrumb)
+        self.help.setText(LABELS[node["type"]])
+        self.help.setToolTip(schema.get("description", "").split("Args:")[0].strip())
+        self.scope.setText(
+            "Available in this step   " + "   ".join(f"${name}" for name in variables)
+        )
+        self.scope.setVisible(bool(variables))
+        advanced = QWidget(self)
+        advanced_form = QFormLayout(advanced)
+        advanced_form.setContentsMargins(0, 8, 0, 8)
+        for name, field in schema["properties"].items():
+            if name in {"type", "children", "body"}:
+                continue
+            value = node.get(name, field.get("default"))
+            editor = self._field_editor(name, value, variables, scans, devices)
+            editor.setToolTip(field.get("description", ""))
+            self.fields[name] = editor
+            label = {
+                "id": "Step name",
+                "name": "Scan type",
+                "device": "Target",
+                "value": "Set to",
+                "variable": "Variable name",
+                "count": "Repeat group",
+                "wait": "",
+                "bind_status": "Status name",
+            }.get(name, name.replace("_", " ").capitalize())
+            target_form = (
+                advanced_form
+                if name in {"bind_status", "arguments", "positional"}
+                or (name == "name" and node["type"] == "scan_call")
+                else self.form
+            )
+            target_form.addRow(label, editor)
+        self.documentation.setVisible(node["type"] == "scan_call")
+        if node["type"] == "scan_call":
+            self._scan_editor(node, variables, scans, advanced_form)
+        self.form.addRow(Section("Advanced settings", advanced, self))
+        for editor in self.fields.values():
+            self._connect_editor(editor)
+        if self.scan_arguments is not None:
+            self.scan_arguments.changed.connect(self.changed.emit)
+
+    def _field_editor(self, name, value, variables, scans, devices):
+        if name == "count":
+            editor = QSpinBox(self)
+            editor.setRange(1, 2_147_483_647)
+            editor.setValue(value)
+            editor.setSuffix(" times")
+        elif name == "wait":
+            editor = QCheckBox("Wait for completion", self)
+            editor.setChecked(value)
+        elif name in {"name", "device"}:
+            editor = QComboBox(self)
+            editor.setEditable(True)
+            editor.addItems(sorted(scans) if name == "name" else devices)
+            editor.setCurrentText(value or "")
+        elif name in {"id", "variable", "bind_status"}:
+            editor = QLineEdit(value or "", self)
+        elif name == "values":
+            editor = ValuesInput(value, variables, self)
+        else:
+            editor = ValueInput(value, variables, self)
+        return editor
+
+    def _scan_editor(self, node, variables, scans, advanced_form):
+        self.fields["name"].currentTextChanged.connect(
+            lambda name: self._scan_metadata(scans.get(name, {}))
+        )
+        self._scan_metadata(scans.get(node["name"], {}))
+        if (
+            isinstance(node.get("arguments"), dict)
+            and not any(key in node["arguments"] for key in ("ref", "param", "type"))
+            and bool(scans)
+            and (not node["name"] or node["name"] in scans)
+        ):
+            self.scan_arguments = ScanDefinitionEditor(node, variables, scans, self.client, self)
+            self.form.addRow(self.scan_arguments)
+            for name in ("arguments", "positional"):
+                self.fields[name].setToolTip(
+                    "Raw JSON inputs, including metadata and structured expressions."
+                )
+            self.raw_arguments = QCheckBox("Edit arguments as raw JSON", self)
+            advanced_form.addRow(self.raw_arguments)
+            self.raw_arguments.toggled.connect(self._argument_mode)
+            self._argument_mode(False)
+            self.fields["name"].currentTextChanged.connect(
+                lambda _: self.raw_arguments.setChecked(True)
+            )
+
+    def _connect_editor(self, editor):
+        if isinstance(editor, (ValueInput, ValuesInput)):
+            editor.changed.connect(self.changed.emit)
+        elif isinstance(editor, QLineEdit):
+            editor.editingFinished.connect(self.changed.emit)
+        elif isinstance(editor, QComboBox):
+            if editor.isEditable():
+                editor.lineEdit().editingFinished.connect(self.changed.emit)
+            editor.activated.connect(lambda *_: self.changed.emit())
+        elif isinstance(editor, QSpinBox):
+            editor.editingFinished.connect(self.changed.emit)
+        elif isinstance(editor, QCheckBox):
+            editor.toggled.connect(lambda *_: self.changed.emit())
+
+    def cleanup(self) -> None:
+        """Release the embedded scan editor before changing the selected component."""
+        if self.scan_arguments is not None:
+            self.scan_arguments.cleanup()
+
+    def _scan_metadata(self, info: dict) -> None:
+        lines = [info.get("doc", ""), "", "Advertised inputs:"]
+        for entry in info.get("signature", []):
+            annotation, _ = ScanInfoAdapter.parse_annotation(entry.get("annotation"))
+            default = entry.get("default", "_empty")
+            requirement = "required" if default == "_empty" else f"default: {default!r}"
+            lines.append(f"{entry['name']} ({annotation}; {entry['kind']}; {requirement})")
+        self.metadata.setPlainText("\n".join(lines))
+
+    def _argument_mode(self, raw: bool) -> None:
+        self.scan_arguments.setEnabled(not raw)
+        for name in ("name", "arguments", "positional"):
+            self.fields[name].setEnabled(raw)
+
+    def values(self) -> dict:
+        """Read component fields without modifying nested children.
+
+        Returns:
+            dict: Edited component preserving its body or children.
+        """
+        node = self.node.copy()
+        for name, editor in self.fields.items():
+            if (
+                name in {"name", "arguments", "positional"}
+                and self.scan_arguments is not None
+                and not self.raw_arguments.isChecked()
+            ):
+                continue
+            if isinstance(editor, (ValueInput, ValuesInput)):
+                value = editor.value()
+            elif isinstance(editor, QComboBox):
+                value = editor.currentText()
+            elif isinstance(editor, QSpinBox):
+                value = editor.value()
+            elif isinstance(editor, QCheckBox):
+                value = editor.isChecked()
+            else:
+                value = editor.text()
+            node[name] = None if name == "bind_status" and not value else value
+        if self.scan_arguments is not None and not self.raw_arguments.isChecked():
+            node["name"], node["positional"], node["arguments"] = self.scan_arguments.values()
+        return node

bec_widgets/widgets/control/scan_scheduler/presentation.py

@@ -0,0 +1,163 @@
+"""Compact step cards and collapsible inspector sections."""
+
+from __future__ import annotations
+
+from qtpy.QtCore import QEvent, QRect, QSize, Qt
+from qtpy.QtGui import QColor, QFont, QPainter, QPen
+from qtpy.QtWidgets import (
+    QApplication,
+    QStyle,
+    QStyledItemDelegate,
+    QToolButton,
+    QVBoxLayout,
+    QWidget,
+)
+
+from bec_widgets.utils.colors import theme_color
+
+# Qt signal descriptors are not understood by pylint.
+# pylint: disable=no-member
+
+LABELS = {
+    "sequence": "Sequence group",
+    "repeat": "Repeat group",
+    "for_each": "For each value",
+    "while": "While condition",
+    "set": "Set device or signal",
+    "scan_call": "Scan",
+    "component_call": "Reusable component",
+}
+
+
+def step_title(node: dict) -> str:
+    """Return the supplied step name or a readable name for a generated identity.
+
+    Args:
+        node (dict): Sequence component.
+
+    Returns:
+        str: Human-readable tree title.
+    """
+    identity = node["id"]
+    if identity.startswith(node["type"] + "_"):
+        if node["type"] == "scan_call" and node.get("name"):
+            return node["name"].replace("_", " ").capitalize()
+        return LABELS[node["type"]]
+    return identity.replace("_", " ").capitalize()
+
+
+class StepDelegate(QStyledItemDelegate):
+    """Paint two-line nested step cards using the application's palette."""
+
+    def sizeHint(self, _option, _index) -> QSize:
+        """Reserve a compact card row.
+
+        Args:
+            _option (QStyleOptionViewItem): Item style options.
+            _index (QModelIndex): Tree item index.
+
+        Returns:
+            QSize: Card dimensions.
+        """
+        return QSize(240, 66)
+
+    def paint(self, painter, option, index) -> None:
+        """Render a step title and its execution behavior.
+
+        Args:
+            painter (QPainter): Active Qt painter.
+            option (QStyleOptionViewItem): Item style options.
+            index (QModelIndex): Tree item index.
+        """
+        painter.save()
+        rect = option.rect.adjusted(2, 3, -8, -3)
+        selected = bool(option.state & QStyle.State_Selected)
+        palette = QApplication.palette()
+        theme = getattr(QApplication.instance(), "theme", None)
+        accent = theme_color(theme, "PRIMARY", palette.highlight().color())
+        background = theme_color(theme, "CARD_BG", palette.window().color())
+        foreground = theme_color(theme, "FG", palette.text().color())
+        border = theme_color(theme, "BORDER", palette.midlight().color())
+        if selected:
+            background = QColor(
+                int(background.red() * 0.9 + accent.red() * 0.1),
+                int(background.green() * 0.9 + accent.green() * 0.1),
+                int(background.blue() * 0.9 + accent.blue() * 0.1),
+            )
+        painter.setRenderHint(QPainter.Antialiasing)
+        painter.setPen(QPen(accent if selected else border, 1))
+        painter.setBrush(background)
+        painter.drawRoundedRect(rect, 5, 5)
+        title = QFont(option.font)
+        title.setBold(True)
+        painter.setFont(title)
+        painter.setPen(foreground)
+        painter.drawText(
+            rect.adjusted(30, 8, -32, -28),
+            Qt.AlignVCenter,
+            option.fontMetrics.elidedText(str(index.data()), Qt.ElideRight, rect.width() - 62),
+        )
+        if index.data(Qt.DecorationRole) is not None:
+            index.data(Qt.DecorationRole).paint(
+                painter, QRect(rect.left() + 10, rect.top() + 11, 14, 14)
+            )
+        painter.drawText(rect.adjusted(rect.width() - 26, 5, -6, -30), Qt.AlignCenter, "⋯")
+        painter.setFont(option.font)
+        muted = QColor(foreground)
+        muted.setAlpha(160)
+        painter.setPen(muted)
+        subtitle = str(index.data(Qt.UserRole + 1) or "")
+        painter.drawText(
+            rect.adjusted(12, 30, -12, -7),
+            Qt.AlignVCenter,
+            option.fontMetrics.elidedText(subtitle, Qt.ElideRight, rect.width() - 24),
+        )
+        painter.restore()
+
+    def editorEvent(self, event, model, option, index) -> bool:
+        """Open step actions from the card's overflow control.
+
+        Args:
+            event (QEvent): Item input event.
+            model (QAbstractItemModel): Tree model.
+            option (QStyleOptionViewItem): Item geometry.
+            index (QModelIndex): Target item.
+
+        Returns:
+            bool: Whether the overflow click was handled.
+        """
+        if (
+            event.type() == QEvent.MouseButtonRelease
+            and event.button() == Qt.LeftButton
+            and event.pos().x() > option.rect.right() - 35
+        ):
+            self.parent().customContextMenuRequested.emit(event.pos())
+            return True
+        return super().editorEvent(event, model, option, index)
+
+
+class Section(QWidget):
+    """Hold optional inspector controls behind a disclosure button.
+
+    Args:
+        title (str): Section label.
+        content (QWidget): Controls to show when expanded.
+        parent (QWidget | None): Owning widget.
+    """
+
+    def __init__(self, title: str, content: QWidget, parent=None) -> None:
+        super().__init__(parent)
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(0, 0, 0, 0)
+        self.toggle = QToolButton(self)
+        self.toggle.setText(title)
+        self.toggle.setCheckable(True)
+        self.toggle.setToolButtonStyle(Qt.ToolButtonTextBesideIcon)
+        self.toggle.setArrowType(Qt.RightArrow)
+        content.setVisible(False)
+        layout.addWidget(self.toggle)
+        layout.addWidget(content)
+        self.toggle.toggled.connect(content.setVisible)
+        self.toggle.toggled.connect(
+            lambda checked: self.toggle.setArrowType(Qt.DownArrow if checked else Qt.RightArrow)
+        )

bec_widgets/widgets/control/scan_scheduler/register_scan_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 bec_widgets.widgets.control.scan_scheduler.scan_scheduler_plugin import ScanSchedulerPlugin
+
+    QPyDesignerCustomWidgetCollection.addCustomWidget(ScanSchedulerPlugin())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

bec_widgets/widgets/control/scan_scheduler/scan_editor.py

@@ -0,0 +1,358 @@
+"""Embed ScanControl, adding scoped references and preserving JSON calling semantics."""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+
+from qtpy.QtCore import QTimer, Signal
+from qtpy.QtGui import QValidator
+from qtpy.QtWidgets import (
+    QCheckBox,
+    QComboBox,
+    QCompleter,
+    QDoubleSpinBox,
+    QLineEdit,
+    QSpinBox,
+    QVBoxLayout,
+    QWidget,
+)
+
+from bec_widgets.utils.widget_io import WidgetIO
+from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
+from bec_widgets.widgets.control.scan_control import ScanControl
+from bec_widgets.widgets.control.scan_control.scan_group_box import ScanLiteralsComboBox
+
+# Qt signal descriptors are not understood by pylint.
+# pylint: disable=no-member
+
+_MISSING = object()
+
+
+class ScanInputValidator(QValidator):
+    """Permit visible variables or literals accepted by the native scan input.
+
+    Args:
+        widget (QWidget): Native input supplying type, bounds and literal validation.
+        variables (list[str]): Loop variables visible at this component.
+        parent (QWidget): Owning input.
+    """
+
+    def __init__(self, widget: QWidget, variables: list[str], parent: QWidget) -> None:
+        super().__init__(parent)
+        self.widget = widget
+        self.variables = {f"${name}" for name in variables}
+
+    def validate(self, text: str, position: int) -> tuple:
+        """Validate a scoped variable or defer to the scan's native literal rules.
+
+        Args:
+            text (str): Current input text.
+            position (int): Cursor position.
+
+        Returns:
+            tuple: Qt validation state, unchanged text and cursor position.
+        """
+        if text in self.variables:
+            return QValidator.State.Acceptable, text, position
+        if text.startswith("$"):
+            state = (
+                QValidator.State.Intermediate
+                if any(name.startswith(text) for name in self.variables)
+                else QValidator.State.Invalid
+            )
+            return state, text, position
+        if isinstance(self.widget, (QSpinBox, QDoubleSpinBox)):
+            return self.widget.validate(text, position)
+        if isinstance(self.widget, QCheckBox):
+            valid = text.lower() in {"true", "false"}
+        elif isinstance(self.widget, DeviceComboBox):
+            valid = self.widget.validate_device(text)
+        elif isinstance(self.widget, ScanLiteralsComboBox):
+            valid = self.widget.findText(text) >= 0
+        else:
+            validator = self.widget.validator() if isinstance(self.widget, QLineEdit) else None
+            if validator is not None:
+                return validator.validate(text, position)
+            valid = True
+        return (
+            (QValidator.State.Acceptable if valid else QValidator.State.Intermediate),
+            text,
+            position,
+        )
+
+
+class BoundScanInput(QLineEdit):
+    """Edit a native scan value or a visible ``$variable`` in one input.
+
+    The validator keeps the native input's literal rules and additionally permits scoped
+    loop variables. Variables are serialized as JSON references, never as Python expressions.
+
+    Args:
+        widget (QWidget): Native scan input supplying literal conversion and validation.
+        value (object): Loaded JSON value or a missing-value sentinel.
+        variables (list[str]): Visible loop bindings.
+        parent (QWidget | None): Owning scan-input tile.
+    """
+
+    changed = Signal()
+
+    def __init__(self, widget: QWidget, value: object, variables: list[str], parent=None) -> None:
+        super().__init__(parent)
+        self.widget = widget
+        self.variables = variables
+        self.setValidator(ScanInputValidator(widget, variables, self))
+        choices = [f"${name}" for name in variables]
+        if isinstance(widget, QComboBox):
+            choices += [widget.itemText(index) for index in range(widget.count())]
+        if choices:
+            self.setCompleter(QCompleter(choices, self))
+        self.setToolTip(
+            widget.toolTip()
+            + (
+                "\nAvailable variables: " + ", ".join(f"${name}" for name in variables)
+                if variables
+                else ""
+            )
+        )
+        if isinstance(value, dict):
+            self.setText(f"${value['ref']}" if set(value) == {"ref"} else json.dumps(value))
+        else:
+            if value is not _MISSING:
+                if isinstance(widget, ScanLiteralsComboBox):
+                    widget.setCurrentText(str(value) if value is not None else "")
+                else:
+                    WidgetIO.set_value(widget, value)
+            self.load_native_value()
+        widget.hide()
+        self.textEdited.connect(lambda *_: self.changed.emit())
+        self.editingFinished.connect(self.changed.emit)
+
+    def load_native_value(self) -> None:
+        """Display a literal restored by ScanControl."""
+        if isinstance(self.widget, (QSpinBox, QDoubleSpinBox)):
+            text = self.widget.text()
+        elif isinstance(self.widget, QCheckBox):
+            text = "true" if self.widget.isChecked() else "false"
+        elif isinstance(self.widget, QComboBox):
+            text = self.widget.currentText()
+        else:
+            text = self.widget.text()
+        self.setText(text)
+
+    def value(self) -> object:
+        """Read a checked literal or scoped JSON reference.
+
+        Returns:
+            object: JSON-compatible scan argument.
+
+        Raises:
+            ValueError: Input is neither a valid literal nor an available variable.
+        """
+        text = self.text()
+        if text.startswith("$"):
+            name = text[1:]
+            if name not in self.variables:
+                raise ValueError(f"Variable {text} is not available in this step")
+            return {"ref": name}
+        if not self.hasAcceptableInput():
+            raise ValueError(f"Invalid {self.widget.arg_name}: {text}")
+        if isinstance(self.widget, (QSpinBox, QDoubleSpinBox)):
+            return self.widget.valueFromText(text)
+        if isinstance(self.widget, QCheckBox):
+            return text.lower() == "true"
+        return text
+
+
+class ScanDefinitionEditor(QWidget):
+    """Use ScanControl for scan selection, argument groups and device bundles.
+
+    The scheduler permits scoped variables in the inputs and translates edits back to
+    their original positional/keyword locations. Opening a node does not supply missing
+    arguments or replace its JSON values with widget defaults.
+
+    Args:
+        node (dict): Scan-call component.
+        variables (list[str]): Visible loop variables.
+        scan_info (dict): Already subscribed scan metadata.
+        client (BECClient): Existing BEC client.
+        parent (QWidget | None): Owning inspector.
+    """
+
+    changed = Signal()
+
+    def __init__(
+        self, node: dict, variables: list[str], scan_info: dict, client, parent=None
+    ) -> None:
+        super().__init__(parent)
+        self.original = deepcopy(node)
+        self.variables = variables
+        self.inputs = {}
+        self.dirty = set()
+        self._name_changed = False
+        self._closed = False
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(0, 0, 0, 0)
+        self.control = ScanControl(
+            parent=self, client=client, scan_info=scan_info, default_scan=node.get("name") or None
+        )
+        self.control.hide_scan_control_buttons = True
+        self.control.scan_control_group.setEnabled(False)
+        if self.control.arg_box is not None and node.get("positional"):
+            defaults, _ = self.control.get_scan_parameters(bec_object=False)
+            values = [
+                defaults[index % len(defaults)] if isinstance(value, dict) else value
+                for index, value in enumerate(node["positional"])
+            ]
+            self.control.arg_box.set_parameters(values)
+        # Metadata can contain expressions; it remains in the scheduler's advanced JSON editor.
+        self.control.hide_metadata = True
+        layout.addWidget(self.control)
+        self.refresh_timer = QTimer(self)
+        self.refresh_timer.setSingleShot(True)
+        self.refresh_timer.timeout.connect(self._refresh_inputs)
+        self._refresh_inputs()
+        self.control.scan_selected.connect(self._selected)
+        self.control.comboBox_scan_selection.activated.connect(self._confirmed)
+        # ScanControl applies this result first; the wrapper then refreshes replaced inputs.
+        # pylint: disable=protected-access
+        self.control._last_scan_parameters_received.connect(self._restored)
+
+    def _restored(self, generation: int, name: str, parameters: tuple[list, dict]) -> None:
+        # Match ScanControl's stale-response guard before reading the restored inputs.
+        # pylint: disable=protected-access
+        if self._closed or generation != self.control._last_scan_fetch_generation:
+            return
+        if name != self.control.current_scan:
+            return
+        positional, keywords = parameters
+        self.original = {
+            "name": name,
+            "positional": deepcopy(positional),
+            "arguments": deepcopy(keywords),
+        }
+        self._name_changed = False
+        self.dirty.clear()
+        for binding in self.inputs.values():
+            binding.load_native_value()
+        self.refresh_timer.start(0)
+        self.changed.emit()
+
+    def _positional_names(self) -> list[str]:
+        signature = self.control.available_scans.get(self.control.current_scan, {}).get(
+            "signature", []
+        )
+        return [
+            entry["name"]
+            for entry in signature
+            if entry["kind"] in {"POSITIONAL_ONLY", "POSITIONAL_OR_KEYWORD"}
+        ]
+
+    def _selected(self, _name):
+        self._name_changed = True
+        self.original = {"name": self.control.current_scan, "positional": [], "arguments": {}}
+        self.inputs.clear()
+        self.dirty.clear()
+        # Native ScanControl restores its inputs after emitting scan_selected.
+        self.refresh_timer.start(0)
+        self.changed.emit()
+
+    def _confirmed(self, _index):
+        if not self.original.get("name"):
+            self._name_changed = True
+            self.changed.emit()
+
+    def _input_changed(self, path):
+        if not self.original.get("name"):
+            self._name_changed = True
+        self.dirty.add(path)
+        self.changed.emit()
+
+    def _refresh_inputs(self):
+        if self._closed:
+            return
+        current = {}
+        if self.control.arg_box is not None:
+            for index, widget in enumerate(self.control.arg_box.widgets):
+                current[("positional", index)] = widget
+            box = self.control.arg_box
+            if not box.property("sequence_bindings_connected"):
+                box.setProperty("sequence_bindings_connected", True)
+                for button in (box.button_add_bundle, box.button_remove_bundle):
+                    button.clicked.connect(self._bundles_changed)
+        for box in self.control.kwarg_boxes:
+            for widget in box.widgets:
+                name = widget.arg_name
+                positional_names = self._positional_names()
+                path = (
+                    ("positional", positional_names.index(name))
+                    if (
+                        self.control.arg_box is None
+                        and name in positional_names
+                        and positional_names.index(name) < len(self.original.get("positional", []))
+                    )
+                    else ("arguments", name)
+                )
+                current[path] = widget
+        existing = {id(binding.widget): binding for binding in self.inputs.values()}
+        self.inputs = {}
+        for path, widget in current.items():
+            if id(widget) in existing:
+                binding = existing[id(widget)]
+                binding.changed.disconnect()
+                binding.changed.connect(lambda path=path: self._input_changed(path))
+                self.inputs[path] = binding
+                continue
+            values = self.original.get(path[0], {} if path[0] == "arguments" else [])
+            value = (
+                values.get(path[1], _MISSING)
+                if isinstance(values, dict)
+                else (values[path[1]] if path[1] < len(values) else _MISSING)
+            )
+            tile = widget.parentWidget()
+            tile.layout().removeWidget(widget)
+            binding = BoundScanInput(widget, value, self.variables, tile)
+            tile.layout().addWidget(binding)
+            binding.changed.connect(lambda path=path: self._input_changed(path))
+            self.inputs[path] = binding
+        self.changed.emit()
+
+    def _bundles_changed(self, *_):
+        self._name_changed = True
+        self.refresh_timer.start(0)
+
+    def values(self) -> tuple[str, list, dict]:
+        """Read native edits while retaining untouched arguments and positional calls.
+
+        Returns:
+            tuple[str, list, dict]: Scan name, positional arguments and keyword arguments.
+        """
+        if self.refresh_timer.isActive():
+            self.refresh_timer.stop()
+            self._refresh_inputs()
+        name = self.control.current_scan if self._name_changed else self.original.get("name", "")
+        positional = deepcopy(self.original.get("positional", []))
+        keywords = deepcopy(self.original.get("arguments", {}))
+        if self._name_changed:
+            positional, defaults = self.control.get_scan_parameters(bec_object=False)
+            keywords.update({key: value for key, value in defaults.items() if key != "metadata"})
+        for path, binding in self.inputs.items():
+            if path not in self.dirty and not self._name_changed:
+                continue
+            value = binding.value()
+            if path[0] == "arguments":
+                keywords[path[1]] = value
+            else:
+                if path[1] >= len(positional):
+                    native, _ = self.control.get_scan_parameters(bec_object=False)
+                    positional.extend(native[len(positional) :])
+                positional[path[1]] = value
+        return name, positional, keywords
+
+    def cleanup(self) -> None:
+        """Release embedded ScanControl resources and pending input refreshes."""
+        if self._closed:
+            return
+        self._closed = True
+        self.refresh_timer.stop()
+        self.control.cleanup()

bec_widgets/widgets/control/scan_scheduler/scan_scheduler.py

@@ -0,0 +1,570 @@
+"""Compose JSON sequences, inspect exports and submit through BEC procedures."""
+
+from __future__ import annotations
+
+import json
+
+# Qt signal descriptors are not understood by pylint.
+# pylint: disable=no-member
+from pathlib import Path
+
+from bec_lib.scan_sequencer import (
+    BECAdapter,
+    DefinitionLibrary,
+    ScanCatalog,
+    Schedule,
+    Sequencer,
+    available_components,
+)
+from bec_qthemes import material_icon
+from qtpy.QtCore import QSignalBlocker, Qt, QTimer
+from qtpy.QtWidgets import (
+    QComboBox,
+    QDialog,
+    QDialogButtonBox,
+    QFileDialog,
+    QHBoxLayout,
+    QLabel,
+    QLineEdit,
+    QMenu,
+    QPlainTextEdit,
+    QPushButton,
+    QScrollArea,
+    QSplitter,
+    QTabWidget,
+    QTreeWidget,
+    QTreeWidgetItem,
+    QVBoxLayout,
+    QWidget,
+)
+
+from bec_widgets import BECWidget, SafeSlot
+
+from .document import SequenceDraft
+from .inspector import ComponentInspector
+from .presentation import LABELS, StepDelegate, step_title
+
+
+class ScanScheduler(BECWidget, QWidget):
+    """Edit nested components and execute validated sequences through procedures.
+
+    Description, Python and JSON representations share the backend's resolver. Execution
+    uses the existing run_sequence procedure without keeping submission records.
+
+    Args:
+        parent (QWidget | None): Owning widget.
+        **kwargs (object): BECWidget connection and configuration arguments.
+    """
+
+    PLUGIN = True
+    ICON_NAME = "account_tree"
+    USER_ACCESS = ["load_json", "document_json", "validate_sequence", "execute_sequence"]
+
+    def __init__(self, parent=None, **kwargs) -> None:
+        super().__init__(parent=parent, **kwargs)
+        self.draft = SequenceDraft()
+        self.schema = available_components()
+        self.catalog = ScanCatalog.from_client(self.client)
+        self._path = ()
+        self._busy = False
+        self._closed = False
+        self._result = {}
+        self.exports = None
+        self.description = None
+        self.python = None
+        self.json = None
+        self.definitions = None
+        self.title = None
+        self.tree = None
+        self.apply_json_button = None
+        self._build_ui()
+        self.edit_timer = QTimer(self)
+        self.edit_timer.setSingleShot(True)
+        self.edit_timer.setInterval(250)
+        self.edit_timer.timeout.connect(self._apply)
+        self.inspector.changed.connect(self.edit_timer.start)
+        self.title.editingFinished.connect(self.edit_timer.start)
+        self.render_timer = QTimer(self)
+        self.render_timer.setSingleShot(True)
+        self.render_timer.setInterval(250)
+        self.render_timer.timeout.connect(self.validate_sequence)
+        self._rebuild_tree()
+
+    def _button(self, text, slot, layout):
+        button = QPushButton(text, self)
+        button.clicked.connect(slot)
+        layout.addWidget(button)
+        return button
+
+    def _build_ui(self) -> None:
+        self.setObjectName("scanScheduler")
+        self.setStyleSheet("""
+            QWidget#scanScheduler { background: palette(base); }
+            QWidget#sequenceOutline { background: palette(alternate-base); border-radius: 5px; }
+            QLabel#sectionHeading, QLabel#brand { font-weight: 600; }
+            QLabel#stepHeading { font-size: 16px; font-weight: 600; margin-top: 12px; }
+            QLabel#breadcrumb { color: palette(placeholder-text); font-size: 11px; }
+            QLabel#scopeChip { background: palette(alternate-base); padding: 8px; border-radius: 4px; }
+            QTreeWidget { background: transparent; border: none; }
+            QLineEdit, QComboBox, QSpinBox { min-height: 24px; }
+            QLineEdit { padding-left: 7px; }
+            QPushButton#primary { background: palette(highlight); color: palette(highlighted-text); padding: 7px 16px; }
+            QTabWidget::pane { border: 1px solid palette(midlight); }
+            QTabBar { qproperty-expanding: false; }
+            QTabBar::tab { min-width: 80px; padding: 8px 12px; margin-right: 5px; }
+            QTreeWidget::item:selected, QTreeWidget::branch:selected { background: transparent; }
+        """)
+        layout = QVBoxLayout(self)
+        layout.setContentsMargins(16, 12, 16, 12)
+        brand = QLabel("BEC   Scan scheduler", self)
+        brand.setObjectName("brand")
+        layout.addWidget(brand)
+        self.editor = QWidget(self)
+        layout.addWidget(self.editor)
+        editor_layout = QVBoxLayout(self.editor)
+        editor_layout.setContentsMargins(0, 12, 0, 0)
+        self._build_toolbar(editor_layout)
+        splitter = QSplitter(self)
+        splitter.setChildrenCollapsible(False)
+        editor_layout.addWidget(splitter, 1)
+        self._build_outline(splitter)
+        scroll = QScrollArea(self)
+        scroll.setWidgetResizable(True)
+        scroll.setFrameShape(QScrollArea.NoFrame)
+        self.inspector = ComponentInspector(self, client=self.client)
+        scroll.setWidget(self.inspector)
+        splitter.addWidget(scroll)
+        splitter.setStretchFactor(0, 1)
+        splitter.setStretchFactor(1, 2)
+        splitter.setSizes([380, 760])
+        self._build_exports(editor_layout)
+        self.message = QLabel("Compose steps and execute the validated sequence.", self)
+        self.message.setWordWrap(True)
+        self.message.setTextFormat(Qt.PlainText)
+        self.message.setObjectName("breadcrumb")
+        layout.addWidget(self.message)
+
+    def _build_toolbar(self, editor_layout):
+        toolbar = QHBoxLayout()
+        self.title = QLineEdit(self.draft.document["id"], self)
+        self.title.setPlaceholderText("Sequence name")
+        toolbar.addWidget(self.title, 1)
+        toolbar.addWidget(QLabel("Draft", self))
+        options = QMenu(self)
+        for text, slot in (
+            ("New sequence", self._new),
+            ("Open sequence…", self._open),
+            ("Save sequence…", self._save),
+            ("Reusable definitions…", self._definitions_dialog),
+        ):
+            options.addAction(text, slot)
+        more = QPushButton("Sequence options  ⋯", self)
+        more.setMenu(options)
+        toolbar.addWidget(more)
+        editor_layout.addLayout(toolbar)
+
+    def _build_outline(self, splitter):
+        outline = QWidget(self)
+        outline.setObjectName("sequenceOutline")
+        outline_layout = QVBoxLayout(outline)
+        outline_layout.setContentsMargins(10, 12, 6, 10)
+        actions = QHBoxLayout()
+        heading = QLabel("Sequence", self)
+        heading.setObjectName("sectionHeading")
+        actions.addWidget(heading)
+        actions.addStretch()
+        self.palette = QComboBox(self)
+        self.palette.addItems(list(self.schema["discriminator"]["mapping"]))
+        self.palette.hide()
+        add_menu = QMenu(self)
+        for kind in self.schema["discriminator"]["mapping"]:
+            add_menu.addAction(
+                LABELS[kind], lambda _checked=False, kind=kind: self._choose_add(kind)
+            )
+        add = QPushButton("+ Add step", self)
+        add.setMenu(add_menu)
+        actions.addWidget(add)
+        outline_layout.addLayout(actions)
+        self.tree = QTreeWidget(self)
+        self.tree.setColumnCount(2)
+        self.tree.setHeaderHidden(True)
+        self.tree.setIndentation(14)
+        self.tree.setItemDelegate(StepDelegate(self.tree))
+        self.tree.setContextMenuPolicy(Qt.CustomContextMenu)
+        self.tree.customContextMenuRequested.connect(self._step_menu)
+        self.tree.currentItemChanged.connect(self._selected)
+        outline_layout.addWidget(self.tree)
+        hint = QLabel("Groups run in order. Repeats apply to their children.", self)
+        hint.setWordWrap(True)
+        hint.setObjectName("breadcrumb")
+        outline_layout.addWidget(hint)
+        splitter.addWidget(outline)
+
+    def _build_exports(self, editor_layout):
+        self.exports = QTabWidget(self)
+        self.exports.tabBar().setExpanding(False)
+        self.exports.setMaximumHeight(210)
+        self.description = self._text_tab(self.exports, "Description", readonly=True)
+        self.python = self._text_tab(self.exports, "Python", readonly=True)
+        self.json = self._text_tab(self.exports, "JSON", readonly=False)
+        self.definitions = QPlainTextEdit(self)
+        self.definitions.setPlainText("[]")
+        self.definitions.hide()
+        editor_layout.addWidget(self.exports)
+        footer = QHBoxLayout()
+        self.apply_json_button = self._button("Apply JSON edits", self._apply_json, footer)
+        self.apply_json_button.hide()
+        self.exports.currentChanged.connect(
+            lambda index: self.apply_json_button.setVisible(index == 2)
+        )
+        footer.addStretch()
+        execute = self._button("Execute", self.execute_sequence, footer)
+        execute.setObjectName("primary")
+        editor_layout.addLayout(footer)
+
+    def _choose_add(self, kind):
+        self.palette.setCurrentText(kind)
+        self._add()
+
+    @SafeSlot(object)
+    def _step_menu(self, point):
+        item = self.tree.itemAt(point)
+        if item is None:
+            return
+        self.tree.setCurrentItem(item)
+        menu = QMenu(self)
+        add = menu.addMenu("Add step")
+        for kind in self.schema["discriminator"]["mapping"]:
+            add.addAction(LABELS[kind], lambda _checked=False, kind=kind: self._choose_add(kind))
+        wrap = menu.addMenu("Wrap in")
+        for kind in ("sequence", "repeat", "for_each", "while"):
+
+            def wrap_step(_checked=False, kind=kind):
+                self.palette.setCurrentText(kind)
+                self._wrap()
+
+            wrap.addAction(LABELS[kind], wrap_step)
+        menu.addSeparator()
+        menu.addAction("Move up", lambda: self._move(-1))
+        menu.addAction("Move down", lambda: self._move(1))
+        menu.addAction("Remove step", self._remove)
+        menu.exec(self.tree.viewport().mapToGlobal(point))
+
+    @SafeSlot()
+    def _definitions_dialog(self):
+        dialog = QDialog(self)
+        dialog.setWindowTitle("Reusable definitions")
+        dialog.resize(700, 500)
+        layout = QVBoxLayout(dialog)
+        layout.addWidget(QLabel("Versioned component definitions (JSON list)", dialog))
+        text = QPlainTextEdit(self.definitions.toPlainText(), dialog)
+        layout.addWidget(text)
+        buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel, dialog)
+        buttons.accepted.connect(dialog.accept)
+        buttons.rejected.connect(dialog.reject)
+        layout.addWidget(buttons)
+        if dialog.exec() == QDialog.Accepted:
+            self.definitions.setPlainText(text.toPlainText())
+            self._sync_json()
+        dialog.deleteLater()
+
+    def _text_tab(self, tabs, title, readonly):
+        text = QPlainTextEdit(self)
+        text.setReadOnly(readonly)
+        tabs.addTab(text, title)
+        return text
+
+    def _schema(self, kind):
+        ref = self.schema["discriminator"]["mapping"][kind].rsplit("/", 1)[-1]
+        return self.schema["$defs"][ref]
+
+    def _show_inspector(self):
+        node = self.draft.node(self._path)
+        self.inspector.display(
+            node,
+            self._schema(node["type"]),
+            self.draft.variables(self._path),
+            self.catalog.entries,
+            sorted(self.client.device_manager.devices),
+            self._breadcrumb(),
+        )
+
+    def _breadcrumb(self):
+        titles = [self.title.text().replace("_", " ")]
+        for length in range(len(self._path) + 1):
+            node = self.draft.node(self._path[:length])
+            if isinstance(node, dict) and node.get("type") != "sequence":
+                titles.append(step_title(node))
+        return " / ".join(titles)
+
+    def _rebuild_tree(self, selected=()):
+        if hasattr(self, "edit_timer"):
+            self.edit_timer.stop()
+        blocker = QSignalBlocker(self.tree)
+        self.tree.clear()
+        target = None
+
+        def add(node, path, parent):
+            nonlocal target
+            kind = node["type"]
+            detail = {
+                "repeat": f"Repeat whole group ×{node.get('count')}",
+                "for_each": f"For each ${node.get('variable')} · {len(node.get('values', []))} values",
+                "set": f"Set {node.get('device')}"
+                + (" and wait" if node.get("wait") else " asynchronously"),
+                "scan_call": node.get("name", "").replace("_", " "),
+                "while": "Check before each iteration",
+                "sequence": "Children execute in order",
+                "component_call": str(node.get("definition", "")),
+            }[kind]
+            item = QTreeWidgetItem(parent, [step_title(node), detail])
+            item.setIcon(
+                0,
+                material_icon(
+                    {
+                        "sequence": "account_tree",
+                        "repeat": "repeat",
+                        "for_each": "format_list_bulleted",
+                        "while": "sync",
+                        "set": "tune",
+                        "scan_call": "filter_center_focus",
+                        "component_call": "extension",
+                    }[kind]
+                ),
+            )
+            item.setData(0, Qt.UserRole + 1, detail)
+            item.setData(0, Qt.UserRole, path)
+            if path == selected:
+                target = item
+            for index, child in enumerate(node.get("children", [])):
+                add(child, path + ("children", index), item)
+            if "body" in node:
+                body = node["body"]
+                if body["type"] == "sequence":
+                    for index, child in enumerate(body["children"]):
+                        add(child, path + ("body", "children", index), item)
+                else:
+                    add(body, path + ("body",), item)
+
+        root = self.draft.document["root"]
+        if root["type"] == "sequence" and root["children"]:
+            for index, child in enumerate(root["children"]):
+                add(child, ("children", index), self.tree)
+        else:
+            add(root, (), self.tree)
+        self.tree.expandAll()
+        self.tree.setCurrentItem(target or self.tree.topLevelItem(0))
+        self._path = tuple(self.tree.currentItem().data(0, Qt.UserRole))
+        del blocker
+        self._show_inspector()
+        self.tree.setColumnHidden(1, True)
+        self.tree.header().setStretchLastSection(True)
+        self._sync_json()
+
+    def _sync_json(self):
+        self.json.setPlainText(json.dumps(self.draft.document, indent=2, ensure_ascii=False))
+        self.description.clear()
+        self.python.clear()
+        self.message.setText("Sequence changed.")
+        if hasattr(self, "render_timer"):
+            self.render_timer.start()
+
+    def _commit(self):
+        node = self.inspector.values()
+        changed = (
+            node != self.draft.node(self._path) or self.draft.document["id"] != self.title.text()
+        )
+        self.draft.replace(self._path, node)
+        self.draft.document["id"] = self.title.text()
+        return changed
+
+    @SafeSlot(object, object)
+    def _selected(self, current, previous):
+        if current is None:
+            return
+        try:
+            changed = self._commit()
+        except (ValueError, TypeError) as exc:
+            blocker = QSignalBlocker(self.tree)
+            self.tree.setCurrentItem(previous)
+            del blocker
+            self.message.setText(f"Step not applied: {exc}")
+            return
+        self._path = tuple(current.data(0, Qt.UserRole))
+        self._show_inspector()
+        if changed:
+            self._sync_json()
+
+    @SafeSlot()
+    def _apply(self):
+        if not self._busy and not self._closed:
+            self._edit(lambda: self._path)
+
+    def _edit(self, action):
+        try:
+            self._commit()
+            selected = action()
+            self._rebuild_tree(selected or ())
+        except (ValueError, TypeError, KeyError) as exc:
+            self.message.setText(str(exc))
+
+    @SafeSlot()
+    def _add(self):
+        self._edit(lambda: self.draft.add(self._path, self.palette.currentText()))
+
+    @SafeSlot()
+    def _wrap(self):
+        def wrap():
+            self.draft.wrap(self._path, self.palette.currentText())
+            return self._path
+
+        self._edit(wrap)
+
+    @SafeSlot()
+    def _remove(self):
+        self._edit(lambda: self.draft.remove(self._path))
+
+    def _move(self, offset):
+        self._edit(lambda: self.draft.move(self._path, offset))
+
+    @SafeSlot()
+    def _new(self):
+        self.draft = SequenceDraft()
+        self.title.setText(self.draft.document["id"])
+        self._rebuild_tree()
+
+    def load_json(self, document: str, definitions: str = "[]") -> None:
+        """Load a structurally valid document and reusable definitions into the editor.
+
+        Args:
+            document (str): Sequence document JSON.
+            definitions (str): JSON list of reusable definitions. Defaults to an empty library.
+        """
+        if self._busy:
+            raise RuntimeError("Wait for the current validation or execution to finish.")
+        schedule = Schedule.model_validate_json(document)
+        DefinitionLibrary(json.loads(definitions))
+        self.draft = SequenceDraft(schedule.model_dump(mode="json"))
+        self.title.setText(schedule.id)
+        self.definitions.setPlainText(definitions)
+        self._rebuild_tree()
+
+    def document_json(self) -> str:
+        """Return the current edited document with structural schema validation.
+
+        Returns:
+            str: Authoritative sequence JSON for saving or submission.
+        """
+        self._commit()
+        return self.draft.validated().model_dump_json(indent=2)
+
+    @SafeSlot()
+    def _apply_json(self):
+        try:
+            self.load_json(self.json.toPlainText(), self.definitions.toPlainText())
+        except (ValueError, TypeError) as exc:
+            self.message.setText(str(exc))
+
+    @SafeSlot()
+    def _open(self):
+        filename, _ = QFileDialog.getOpenFileName(self, "Open sequence", filter="JSON (*.json)")
+        if filename:
+            try:
+                self.load_json(
+                    Path(filename).read_text(encoding="utf-8"), self.definitions.toPlainText()
+                )
+            except (OSError, ValueError) as exc:
+                self.message.setText(str(exc))
+
+    @SafeSlot()
+    def _save(self):
+        filename, _ = QFileDialog.getSaveFileName(self, "Save sequence", filter="JSON (*.json)")
+        if filename:
+            try:
+                Path(filename).write_text(self.document_json(), encoding="utf-8")
+            except (OSError, ValueError) as exc:
+                self.message.setText(str(exc))
+
+    @SafeSlot()
+    def validate_sequence(self) -> None:
+        """Validate the entire draft and refresh all exports on a background task."""
+        self._start(False)
+
+    @SafeSlot()
+    def execute_sequence(self) -> None:
+        """Validate and execute the current sequence through the existing procedure API."""
+        self._start(True)
+
+    def _start(self, execute):
+        self.edit_timer.stop()
+        self.render_timer.stop()
+        if self._busy:
+            return
+        try:
+            payload = self.document_json()
+            definitions = json.loads(self.definitions.toPlainText())
+        except (ValueError, TypeError) as exc:
+            self.message.setText(str(exc))
+            return
+        self.description.clear()
+        self.python.clear()
+        self.json.setPlainText(payload)
+        self._busy = True
+        self.editor.setEnabled(False)
+        self.message.setText("Validating and executing…" if execute else "Validating…")
+        self._result = {}
+        client = self.client
+        result = self._result
+
+        def prepare():
+            catalog = ScanCatalog.from_client(client)
+            library = DefinitionLibrary(definitions)
+            sequencer = Sequencer(catalog, library)
+            sequencer.validate(payload, BECAdapter(client) if execute else None)
+            description = sequencer.export(payload, format="description")
+            python = sequencer.export(payload, format="python")
+            _, closure = library.resolve(Schedule.model_validate_json(payload).root)
+            result.update(description=description, python=python, payload=payload)
+            if execute:
+                client.proc.request_new(
+                    "run_sequence",
+                    ((payload, [entry.model_dump(mode="json") for entry in closure]), {}),
+                )
+                result["executed"] = True
+
+        self.submit_task(prepare, on_complete=self._prepared, on_failed=self._failed)
+
+    @SafeSlot()
+    def _prepared(self):
+        if self._closed:
+            return
+        self._busy = False
+        self.editor.setEnabled(True)
+        self.description.setPlainText(self._result["description"])
+        self.python.setPlainText(self._result["python"])
+        self.json.setPlainText(self._result["payload"])
+        self.message.setText(
+            "Execution requested through run_sequence."
+            if self._result.get("executed")
+            else "Sequence valid."
+        )
+
+    @SafeSlot(str)
+    def _failed(self, error):
+        if self._closed:
+            return
+        self._busy = False
+        self.editor.setEnabled(True)
+        self.message.setText(error.strip().splitlines()[-1])
+        self.message.setToolTip(error)
+
+    def cleanup(self) -> None:
+        """Stop pending edits and disconnect widget resources."""
+        self._closed = True
+        if hasattr(self, "edit_timer"):
+            self.edit_timer.stop()
+        if hasattr(self, "render_timer"):
+            self.render_timer.stop()
+        if hasattr(self, "inspector"):
+            self.inspector.cleanup()
+        super().cleanup()

bec_widgets/widgets/control/scan_scheduler/scan_scheduler.pyproject

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

bec_widgets/widgets/control/scan_scheduler/scan_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 qtpy.QtDesigner import QDesignerCustomWidgetInterface
+from qtpy.QtWidgets import QWidget
+
+from bec_widgets.utils.bec_designer import designer_material_icon
+from bec_widgets.widgets.control.scan_scheduler.scan_scheduler import ScanScheduler
+
+DOM_XML = """
+<ui language='c++'>
+    <widget class='ScanScheduler' name='scan_scheduler'>
+    </widget>
+</ui>
+"""
+
+
+class ScanSchedulerPlugin(QDesignerCustomWidgetInterface):  # pragma: no cover
+    def __init__(self):
+        super().__init__()
+        self._form_editor = None
+
+    def createWidget(self, parent):
+        if parent is None:
+            return QWidget()
+        t = ScanScheduler(parent)
+        return t
+
+    def domXml(self):
+        return DOM_XML
+
+    def group(self):
+        return ""
+
+    def icon(self):
+        return designer_material_icon(ScanScheduler.ICON_NAME)
+
+    def includeFile(self):
+        return "scan_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 "ScanScheduler"
+
+    def toolTip(self):
+        return "Edit nested components and execute validated sequences through procedures."
+
+    def whatsThis(self):
+        return self.toolTip()

docs/scan_scheduler/README.md

@@ -0,0 +1,62 @@
+# Scan scheduler
+
+The scheduler composes nested BEC sequence components and executes their JSON to the existing
+`run_sequence` procedure. It requires the BEC `feature/scan_sequencer` branch, installed in both
+this GUI environment and the procedure worker environment. There is no new execution service.
+
+Open **Scan Scheduler** in `bec-app`, add a ScanScheduler through the GUI server/Designer, or run:
+
+```bash
+python -m bec_widgets.examples.scan_scheduler
+```
+
+Opening the example does not move hardware or acquire scans. Its `samx` and `time_scan` inputs
+must be available on your deployment before validation/execution succeeds.
+
+## Compose and inspect
+
+- Select a sequence or loop, choose a component and click **Add step**. A selected sequence child
+  inserts a sibling after that child. **Wrap** surrounds the selected step with a sequence or loop.
+- Edit the selected component in the inspector. Changes save automatically. Repeat counts must
+  be positive. The card menu provides wrap, reorder and remove actions.
+- Set targets accept dotted device/signal names. **Wait for completion** controls native blocking
+  set/move behavior. Visible for_each variables appear in the **Variable** selector.
+- Scan definitions use the existing **ScanControl** widget: its scan selector, argument groups,
+  device bundles, input bounds, units, tooltips and **Restore last** behavior. The scheduler hides
+  its execution buttons; composing a step does not submit a scan. Inputs accept a visible loop variable directly, for example `$exposure`.
+  Autocompletion offers the current scope; numeric literals retain their native bounds and units. Existing positional inputs remain positional, and untouched missing
+  arguments remain missing so backend validation can reject them.
+- **Advanced settings → Edit arguments as raw JSON** supports metadata, whole argument-mapping
+  references and structured expressions. For example `{"ref": "pol"}`. Use
+  `{"type": "literal", "value": ...}` to protect dictionaries resembling reserved expressions.
+- **Description**, **Python** and **JSON** update automatically after editing. Description is
+  selected first. Validation covers the complete tree, references and definitions. Execution also
+  checks live scan/device inputs before requesting the procedure. Invalid sequences cannot execute.
+- **JSON → Apply JSON** imports edits made to the document JSON. **Sequence options → Open/Save**
+  uses standalone sequence documents. **Reusable definitions** accepts a JSON list of versioned
+  backend definitions; save/share that list separately alongside a sequence that needs it.
+  This prototype does not provide a reusable-definition authoring form.
+
+## Execute
+
+Choose **Execute** to validate the sequence and request the existing `run_sequence` procedure.
+Validation and the request run off the Qt thread. The request contains the current JSON document
+and the transitive reusable definitions it needs. The worker independently validates these inputs
+before using native IPython scan/device calls. Scan implementations must be available in that worker.
+
+The widget does not keep submission records, queue views or history. Procedure execution and
+interruption remain owned by BEC. Closing this editor does not interrupt a running procedure.
+
+## Programmatic use
+
+```python
+from bec_widgets.widgets.control.scan_scheduler import ScanScheduler
+
+scheduler = ScanScheduler(client=bec)
+scheduler.load_json(sequence_json, definitions_json)  # definitions defaults to "[]"
+scheduler.show()
+scheduler.validate_sequence()  # optional: refresh immediately instead of waiting for automatic updates
+# scheduler.execute_sequence()  # explicit execution action
+```
+
+![Scheduler with native ScanControl and description](scheduler-revised.png)

tests/unit_tests/scan_scheduler/scan_info.json

@@ -0,0 +1,472 @@
+{
+  "time_scan": {
+    "class": "TimeScan",
+    "base_class": "ScanBaseV4",
+    "is_scan": true,
+    "is_internal": false,
+    "arg_input": {},
+    "required_kwargs": [],
+    "arg_bundle_size": {
+      "bundle": 0,
+      "min": null,
+      "max": null
+    },
+    "signature": [
+      {
+        "name": "points",
+        "kind": "POSITIONAL_OR_KEYWORD",
+        "default": "_empty",
+        "annotation": {
+          "Annotated": {
+            "type": "int",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": null,
+                "description": null,
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": null,
+                "reference_units": null,
+                "gt": null,
+                "ge": null,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "interval",
+        "kind": "POSITIONAL_OR_KEYWORD",
+        "default": "_empty",
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Interval",
+                "description": null,
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "exp_time",
+        "kind": "POSITIONAL_OR_KEYWORD",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Exposure Time",
+                "description": "Exposure time in seconds",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "settling_time",
+        "kind": "POSITIONAL_OR_KEYWORD",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Settling Time",
+                "description": "Time to settle before trigger and readout.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "kwargs",
+        "kind": "VAR_KEYWORD",
+        "default": "_empty",
+        "annotation": "_empty"
+      }
+    ],
+    "doc": "Trigger and readout devices at a fixed interval.\nNote that the interval time cannot be less than the exposure time.\nThe effective sleep time between points is:\n    sleep_time = max(interval - exp_time, 0)\n\nArgs:\n    points (int): points.\n    interval (float [s]): interval. s\n    exp_time (float [s]): Exposure time in seconds. Default: 0 s\n    settling_time (float [s]): Time to settle before trigger and readout. Default: 0 s\n\nReturns:\n    ScanReport\n\nExamples:\n    Minimum:\n        >>> scans.time_scan(1, 1.0)\n    Full:\n        >>> scans.time_scan(1, 1.0, exp_time=0, settling_time=0)",
+    "gui_visibility": {
+      "Scan Parameters": [
+        "points",
+        "interval",
+        "exp_time",
+        "settling_time"
+      ]
+    }
+  },
+  "line_scan": {
+    "class": "LineScan",
+    "base_class": "ScanBaseV4",
+    "is_scan": true,
+    "is_internal": false,
+    "arg_input": {
+      "device": "DeviceBase",
+      "start": {
+        "Annotated": {
+          "type": "float",
+          "metadata": {
+            "ScanArgument": {
+              "display_name": "Start Position",
+              "description": null,
+              "tooltip": null,
+              "expert": false,
+              "hidden": false,
+              "example": null,
+              "units": null,
+              "reference_units": "device",
+              "gt": null,
+              "ge": null,
+              "lt": null,
+              "le": null,
+              "reference_limits": null,
+              "precision": null,
+              "alternative_group": null
+            }
+          }
+        }
+      },
+      "stop": {
+        "Annotated": {
+          "type": "float",
+          "metadata": {
+            "ScanArgument": {
+              "display_name": "Stop Position",
+              "description": null,
+              "tooltip": null,
+              "expert": false,
+              "hidden": false,
+              "example": null,
+              "units": null,
+              "reference_units": "device",
+              "gt": null,
+              "ge": null,
+              "lt": null,
+              "le": null,
+              "reference_limits": null,
+              "precision": null,
+              "alternative_group": null
+            }
+          }
+        }
+      }
+    },
+    "required_kwargs": [],
+    "arg_bundle_size": {
+      "bundle": 3,
+      "min": 1,
+      "max": null
+    },
+    "signature": [
+      {
+        "name": "args",
+        "kind": "VAR_POSITIONAL",
+        "default": "_empty",
+        "annotation": "_empty"
+      },
+      {
+        "name": "steps",
+        "kind": "KEYWORD_ONLY",
+        "default": "_empty",
+        "annotation": {
+          "Annotated": {
+            "type": "int",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Number of Steps",
+                "description": null,
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": null,
+                "reference_units": null,
+                "gt": 0.0,
+                "ge": null,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "relative",
+        "kind": "KEYWORD_ONLY",
+        "default": "_empty",
+        "annotation": {
+          "Annotated": {
+            "type": "bool",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Relative",
+                "description": "Whether the positions are relative to the current position",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": null,
+                "reference_units": null,
+                "gt": null,
+                "ge": null,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "exp_time",
+        "kind": "KEYWORD_ONLY",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Exposure Time",
+                "description": "Exposure time in seconds",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "frames_per_trigger",
+        "kind": "KEYWORD_ONLY",
+        "default": 1,
+        "annotation": {
+          "Annotated": {
+            "type": "int",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Frames per Trigger",
+                "description": "Number of frames per trigger for devices that support configurable frame counts per trigger.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": null,
+                "reference_units": null,
+                "gt": null,
+                "ge": 1.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "settling_time",
+        "kind": "KEYWORD_ONLY",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Settling Time",
+                "description": "Time to settle before trigger and readout.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "settling_time_after_trigger",
+        "kind": "KEYWORD_ONLY",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Settling Time After Trigger",
+                "description": "Time to settle after the trigger but before the readout.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "readout_time",
+        "kind": "KEYWORD_ONLY",
+        "default": 0,
+        "annotation": {
+          "Annotated": {
+            "type": "float",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Readout Time",
+                "description": "Configuration for devices that support configurable readout times.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": "s",
+                "reference_units": null,
+                "gt": null,
+                "ge": 0.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "burst_at_each_point",
+        "kind": "KEYWORD_ONLY",
+        "default": 1,
+        "annotation": {
+          "Annotated": {
+            "type": "int",
+            "metadata": {
+              "ScanArgument": {
+                "display_name": "Burst at Each Point",
+                "description": "Number of triggers and readouts at each point.",
+                "tooltip": null,
+                "expert": false,
+                "hidden": false,
+                "example": null,
+                "units": null,
+                "reference_units": null,
+                "gt": null,
+                "ge": 1.0,
+                "lt": null,
+                "le": null,
+                "reference_limits": null,
+                "precision": null,
+                "alternative_group": null
+              }
+            }
+          }
+        }
+      },
+      {
+        "name": "kwargs",
+        "kind": "VAR_KEYWORD",
+        "default": "_empty",
+        "annotation": "_empty"
+      }
+    ],
+    "doc": "A line scan for one or more motors.\n\nArgs:\n    *args (device: DeviceBase, start: float, stop: float): repeated scan argument bundles.\n    steps (int): steps.\n    relative (bool): Whether the positions are relative to the current position.\n    exp_time (float [s]): Exposure time in seconds. Default: 0 s\n    frames_per_trigger (int): Number of frames per trigger for devices that support configurable frame counts per trigger. Default: 1\n    settling_time (float [s]): Time to settle before trigger and readout. Default: 0 s\n    settling_time_after_trigger (float [s]): Time to settle after the trigger but before the readout. Default: 0 s\n    readout_time (float [s]): Configuration for devices that support configurable readout times. Default: 0 s\n    burst_at_each_point (int): Number of triggers and readouts at each point. Default: 1\n\nReturns:\n    ScanReport\n\nExamples:\n    Minimum:\n        >>> scans.line_scan(dev.device1, -1.0, 1.0, steps=10, relative=False)\n    Full:\n        >>> scans.line_scan(dev.device1, -1.0, 1.0, steps=10, relative=False, exp_time=0, frames_per_trigger=1, settling_time=0, settling_time_after_trigger=0, readout_time=0, burst_at_each_point=1)",
+    "gui_visibility": {
+      "Movement Parameters": [
+        "steps",
+        "relative"
+      ],
+      "Acquisition Parameters": [
+        "exp_time",
+        "frames_per_trigger",
+        "settling_time",
+        "settling_time_after_trigger",
+        "readout_time",
+        "burst_at_each_point"
+      ]
+    }
+  }
+}

tests/unit_tests/scan_scheduler/test_scan_scheduler.py

@@ -0,0 +1,476 @@
+"""Scheduler composition and authoritative procedure integration tests."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from threading import Event, get_ident
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from bec_widgets.tests.utils import create_widget
+from bec_widgets.widgets.control.scan_scheduler import ScanScheduler
+from bec_widgets.widgets.control.scan_scheduler.document import SequenceDraft
+from bec_widgets.widgets.control.scan_scheduler.inspector import ValueInput
+
+
+@pytest.fixture
+def scheduler(qtbot, mocked_client):
+    info = json.loads(Path(__file__).with_name("scan_info.json").read_text())
+    mocked_client.scans._available_scans = {
+        name: SimpleNamespace(scan_info=entry) for name, entry in info.items()
+    }
+    mocked_client.proc = MagicMock()
+    return create_widget(qtbot, ScanScheduler, client=mocked_client)
+
+
+@pytest.fixture
+def sequence():
+    return {
+        "schema_version": 1,
+        "type": "sequence",
+        "id": "alternating_positions",
+        "revision": 1,
+        "root": {
+            "id": "repeat",
+            "type": "repeat",
+            "count": 2,
+            "body": {
+                "id": "positions",
+                "type": "for_each",
+                "variable": "position",
+                "values": [1, -1],
+                "body": {
+                    "id": "steps",
+                    "type": "sequence",
+                    "children": [
+                        {
+                            "id": "move",
+                            "type": "set",
+                            "device": "samx",
+                            "value": {"ref": "position"},
+                            "wait": True,
+                        },
+                        {
+                            "id": "scan",
+                            "type": "scan_call",
+                            "name": "time_scan",
+                            "positional": [1, 0.1],
+                            "arguments": {},
+                        },
+                    ],
+                },
+            },
+        },
+    }
+
+
+def test_tree_and_scope(scheduler, sequence):
+    scheduler.load_json(json.dumps(sequence))
+    assert scheduler.exports.tabText(0) == "Description"
+    scheduler._rebuild_tree(("body", "body", "children", 0))
+    assert scheduler.inspector.fields["value"].mode.currentData() == "position"
+    assert "position" in scheduler.inspector.scope.text()
+    assert "and wait" in scheduler.tree.currentItem().text(1)
+    assert json.loads(scheduler.document_json())["root"] == scheduler.draft.document["root"]
+
+
+def test_validate_exports_in_worker(scheduler, sequence, qtbot):
+    scheduler.load_json(json.dumps(sequence))
+    scheduler.validate_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert "Sequence valid" in scheduler.message.text()
+    assert "wait" in scheduler.description.toPlainText()
+    assert "time_scan" in scheduler.python.toPlainText()
+    assert scheduler.client.proc.request_new.call_count == 0
+
+
+def test_invalid_scan_never_submitted(scheduler, sequence, qtbot):
+    sequence["root"]["body"]["body"]["children"][1]["positional"] = []
+    scheduler.load_json(json.dumps(sequence))
+    scheduler.execute_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert scheduler.client.proc.request_new.call_count == 0
+    assert "points" in scheduler.message.text()
+
+
+def test_execute_uses_procedure_without_retaining_records(scheduler, sequence, qtbot):
+    scheduler.load_json(json.dumps(sequence))
+    scheduler.execute_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    request = scheduler.client.proc.request_new.call_args
+    assert request.args[0] == "run_sequence"
+    document, _definitions = request.args[1][0]
+    assert json.loads(document)["revision"] == 1
+    assert scheduler.draft.document["revision"] == 1
+    assert not hasattr(scheduler, "records")
+    assert not hasattr(scheduler, "queue_table")
+    assert not hasattr(scheduler, "history_table")
+
+
+def test_submission_does_not_block_event_loop(scheduler, sequence, qtbot):
+    started, release = Event(), Event()
+    threads = []
+
+    def request(*args, **kwargs):
+        threads.append(get_ident())
+        started.set()
+        assert release.wait(5)
+        return SimpleNamespace(done=True, state=SimpleNamespace(value="SUCCESS"), error=None)
+
+    scheduler.client.proc.request_new.side_effect = request
+    scheduler.load_json(json.dumps(sequence))
+    scheduler.execute_sequence()
+    try:
+        qtbot.waitUntil(started.is_set)
+        assert scheduler._busy
+        assert threads == [threads[0]] and threads[0] != get_ident()
+    finally:
+        release.set()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+
+
+def test_json_errors_preserve_draft(scheduler, sequence):
+    scheduler.load_json(json.dumps(sequence))
+    original = scheduler.document_json()
+    scheduler.json.setPlainText('{"type":"scan"}')
+    scheduler._apply_json()
+    assert scheduler.document_json() == original
+
+
+def test_edit_add_wrap_move_remove(scheduler):
+    scheduler.palette.setCurrentText("set")
+    scheduler._add()
+    scheduler.inspector.fields["device"].setCurrentText("samx.velocity")
+    scheduler.inspector.fields["value"].value_edit.setText("2")
+    scheduler._apply()
+    scheduler.palette.setCurrentText("repeat")
+    scheduler._wrap()
+    assert scheduler.draft.node(scheduler._path)["count"] == 1
+    assert scheduler.inspector.fields["count"].minimum() == 1
+    scheduler._remove()
+    assert scheduler.draft.document["root"]["children"] == []
+
+
+def test_document_nested_scope_and_reordering(sequence):
+    draft = SequenceDraft(sequence)
+    path = ("body", "body", "children", 0)
+    assert draft.variables(path) == ["position"]
+    assert draft.variables(("body",)) == []
+    assert draft.move(path, 1)[-1] == 1
+    draft.remove(("body", "body"))
+    assert draft.node(("body", "body"))["children"] == []
+
+
+def test_value_reference_and_safe_literal(qtbot):
+    widget = ValueInput({"ref": "pol"}, ["pol"])
+    qtbot.addWidget(widget)
+    assert widget.value() == {"ref": "pol"}
+    widget.mode.setCurrentIndex(0)
+    widget.value_edit.setText(json.dumps("'\n__import__('os')"))
+    assert widget.value() == "'\n__import__('os')"
+
+
+def test_close_stops_pending_edits(scheduler):
+    scheduler.render_timer.start()
+    scheduler.close()
+    assert not scheduler.render_timer.isActive()
+
+
+def test_named_scan_arguments_preserve_semantics(scheduler, sequence, qtbot):
+    sequence["root"]["body"]["values"] = [0.1, 0.2]
+    scheduler.load_json(json.dumps(sequence))
+    path = ("body", "body", "children", 1)
+    scheduler._rebuild_tree(path)
+    inputs = scheduler.inspector.scan_arguments
+    assert inputs.values() == ("time_scan", [1, 0.1], {})
+    # Bind exposure time to the visible variable without rewriting positional arguments.
+    editor = inputs.inputs[("arguments", "exp_time")]
+    editor.setText("$position")
+    editor.editingFinished.emit()
+    scheduler._apply()
+    scan = scheduler.draft.node(path)
+    assert scan["positional"] == [1, 0.1]
+    assert scan["arguments"]["exp_time"] == {"ref": "position"}
+    scheduler.validate_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert "time_scan" in scheduler.description.toPlainText(), scheduler.message.text()
+
+
+def test_execution_passes_transitive_definitions(scheduler, qtbot):
+    definitions = [
+        {
+            "id": "inner",
+            "version": 1,
+            "parameters": {"type": "object", "additionalProperties": False},
+            "body": {
+                "id": "scan",
+                "type": "scan_call",
+                "name": "time_scan",
+                "positional": [1, 0.1],
+            },
+        },
+        {
+            "id": "outer",
+            "version": 1,
+            "body": {
+                "id": "call",
+                "type": "component_call",
+                "definition": {"id": "inner", "version": 1},
+            },
+        },
+        {"id": "unused", "version": 1, "body": {"id": "empty", "type": "sequence", "children": []}},
+    ]
+    document = {
+        "id": "reusable",
+        "revision": 1,
+        "root": {
+            "id": "call",
+            "type": "component_call",
+            "definition": {"id": "outer", "version": 1},
+        },
+    }
+    scheduler.client.proc.request_new.return_value = SimpleNamespace(
+        done=False, state=SimpleNamespace(value="RUNNING"), error=None
+    )
+    scheduler.load_json(json.dumps(document), json.dumps(definitions))
+    scheduler.execute_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    request = scheduler.client.proc.request_new.call_args
+    assert request is not None, scheduler.message.text()
+    definitions = request.args[1][0][1]
+    assert [entry["id"] for entry in definitions] == ["inner", "outer"]
+
+
+def test_invalid_reference_rejected_before_submission(scheduler, sequence, qtbot):
+    sequence["root"]["body"]["body"]["children"][0]["value"] = {"ref": "missing"}
+    scheduler.load_json(json.dumps(sequence))
+    scheduler.execute_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert "missing" in scheduler.message.text()
+    scheduler.client.proc.request_new.assert_not_called()
+
+
+def test_repeat_zero_and_unknown_types_rejected(scheduler, sequence):
+    sequence["root"]["count"] = 0
+    with pytest.raises(ValueError):
+        scheduler.load_json(json.dumps(sequence))
+    sequence["root"]["count"] = 1
+    sequence["type"] = "scan"
+    with pytest.raises(ValueError):
+        scheduler.load_json(json.dumps(sequence))
+
+
+def test_description_capture(scheduler, qtbot):
+    import os
+
+    from bec_widgets.examples.scan_scheduler import example_sequence
+
+    scheduler.load_json(example_sequence().model_dump_json())
+    scheduler.resize(1400, 1000)
+    scheduler._rebuild_tree(("body", "body", "children", 0))
+    scheduler.show()
+    scheduler.validate_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert scheduler.description.toPlainText()
+    if filename := os.environ.get("BEC_SCHEDULER_SCREENSHOT"):
+        qtbot.wait(100)
+        assert scheduler.grab().save(filename)
+    if filename := os.environ.get("BEC_SCHEDULER_PYTHON_SCREENSHOT"):
+        scheduler.exports.setCurrentIndex(1)
+        qtbot.wait(100)
+        assert scheduler.grab().save(filename)
+
+
+def test_nested_scan_inspector_capture(scheduler, qtbot):
+    import os
+
+    def scan(identity, exposure):
+        return {
+            "id": identity,
+            "type": "scan_call",
+            "name": "line_scan",
+            "positional": ["samx", -1.0, 1.0],
+            "arguments": {
+                "steps": 30,
+                "relative": False,
+                "exp_time": exposure,
+                "metadata": {"sample_name": "sample1"},
+            },
+        }
+
+    document = {
+        "id": "Position & exposure",
+        "revision": 1,
+        "root": {
+            "id": "root",
+            "type": "sequence",
+            "children": [
+                {
+                    "id": "Position cycle",
+                    "type": "repeat",
+                    "count": 3,
+                    "body": {
+                        "id": "Position sweep",
+                        "type": "for_each",
+                        "variable": "position",
+                        "values": [-1, 1],
+                        "body": {
+                            "id": "position_steps",
+                            "type": "sequence",
+                            "children": [
+                                {
+                                    "id": "Set position",
+                                    "type": "set",
+                                    "device": "samx",
+                                    "value": {"ref": "position"},
+                                    "wait": True,
+                                },
+                                {
+                                    "id": "Repeat acquisition",
+                                    "type": "repeat",
+                                    "count": 2,
+                                    "body": scan("Position line", 0.1),
+                                },
+                            ],
+                        },
+                    },
+                },
+                {
+                    "id": "Exposure series",
+                    "type": "for_each",
+                    "variable": "exposure",
+                    "values": [0.1, 0.2],
+                    "body": {
+                        "id": "Repeat exposure acquisition",
+                        "type": "repeat",
+                        "count": 2,
+                        "body": scan("Exposure line", {"ref": "exposure"}),
+                    },
+                },
+            ],
+        },
+    }
+    scheduler.load_json(json.dumps(document))
+    scheduler.resize(1250, 1050)
+    scheduler._rebuild_tree(("children", 1, "body", "body"))
+    scheduler.show()
+    scheduler.validate_sequence()
+    qtbot.waitUntil(lambda: not scheduler._busy)
+    assert scheduler.description.toPlainText(), scheduler.message.text()
+    assert "$exposure" in scheduler.inspector.scope.text()
+    if filename := os.environ.get("BEC_SCHEDULER_REVISED_SCREENSHOT"):
+        qtbot.wait(100)
+        assert scheduler.grab().save(filename)
+
+
+def test_scan_control_is_embedded_without_fetching_metadata(scheduler, sequence, monkeypatch):
+    from bec_widgets.widgets.control.scan_control import ScanControl
+
+    def unexpected_fetch(*args, **kwargs):
+        raise AssertionError("Scan metadata is already subscribed")
+
+    monkeypatch.setattr(scheduler.client.connector, "get", unexpected_fetch)
+    scheduler.load_json(json.dumps(sequence))
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    editor = scheduler.inspector.scan_arguments
+    assert isinstance(editor.control, ScanControl)
+    assert editor.control.hide_scan_control_buttons
+    assert not editor.control.scan_control_group.isEnabled()
+    assert editor.values() == ("time_scan", [1, 0.1], {})
+
+
+def test_native_restore_refreshes_bindings(scheduler, sequence, qtbot):
+    scheduler.load_json(json.dumps(sequence))
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    editor = scheduler.inspector.scan_arguments
+    binding = editor.inputs[("arguments", "exp_time")]
+    binding.setText("$position")
+    binding.editingFinished.emit()
+    control = editor.control
+    control._last_scan_parameters_received.emit(
+        control._last_scan_fetch_generation, "time_scan", ([], {"points": 5, "interval": 0.2})
+    )
+    qtbot.waitUntil(lambda: not editor.refresh_timer.isActive())
+    assert editor.values() == ("time_scan", [], {"points": 5, "interval": 0.2})
+    assert binding.text() == "0.00 s"
+
+
+def test_missing_scan_arguments_stay_missing(scheduler, sequence):
+    scan = sequence["root"]["body"]["body"]["children"][1]
+    scan["positional"] = []
+    scheduler.load_json(json.dumps(sequence))
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    assert scheduler.inspector.scan_arguments.values() == ("time_scan", [], {})
+
+
+def test_new_scan_uses_native_selector(scheduler, sequence):
+    scan = sequence["root"]["body"]["body"]["children"][1]
+    scheduler.load_json(json.dumps(sequence))
+    scan = scheduler.draft.document["root"]["body"]["body"]["children"][1]
+    scan["name"] = ""
+    scan["positional"] = []
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    editor = scheduler.inspector.scan_arguments
+    assert editor is not None
+    editor.control.current_scan = "time_scan"
+    editor._confirmed(0)
+    assert editor.values()[0] == "time_scan"
+
+
+def test_native_scan_device_bundles_preserve_positional_inputs(scheduler, sequence, qtbot):
+    node = sequence["root"]["body"]["body"]["children"][1]
+    node.update(
+        name="line_scan",
+        positional=["samx", -1, 1, "samy", -2, 2],
+        arguments={"steps": 3, "relative": False},
+    )
+    scheduler.load_json(json.dumps(sequence))
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    editor = scheduler.inspector.scan_arguments
+    assert editor.control.arg_box.count_arg_rows() == 2
+    assert editor.values() == ("line_scan", node["positional"], node["arguments"])
+    editor.control.arg_box.button_remove_bundle.click()
+    qtbot.waitUntil(lambda: not editor.refresh_timer.isActive())
+    name, positional, keywords = editor.values()
+    assert name == "line_scan"
+    assert positional == ["samx", -1, 1]
+    assert keywords["steps"] == 3
+
+
+def test_direct_variable_input_respects_scope_and_literal_bounds(scheduler, sequence):
+    from qtpy.QtGui import QValidator
+
+    sequence["root"]["body"]["body"]["children"][1].update(
+        name="line_scan", positional=["samx", -1, 1], arguments={"steps": 3, "relative": False}
+    )
+    scheduler.load_json(json.dumps(sequence))
+    scheduler._rebuild_tree(("body", "body", "children", 1))
+    field = scheduler.inspector.scan_arguments.inputs[("arguments", "steps")]
+    validator = field.validator()
+    assert validator.validate("$position", 9)[0] == QValidator.State.Acceptable
+    field.setText("$position")
+    assert field.value() == {"ref": "position"}
+    field.setText("$missing")
+    with pytest.raises(ValueError, match="not available"):
+        field.value()
+    field.setText("0")
+    with pytest.raises(ValueError, match="Invalid steps"):
+        field.value()
+    field.setText("5")
+    assert field.value() == 5
+
+
+def test_description_updates_without_preview_action(scheduler, sequence, qtbot):
+    from qtpy.QtWidgets import QPushButton
+
+    scheduler.load_json(json.dumps(sequence))
+    qtbot.waitUntil(lambda: bool(scheduler.description.toPlainText()))
+    assert scheduler.exports.count() == 3
+    buttons = [button.text() for button in scheduler.findChildren(QPushButton)]
+    assert "Execute" in buttons
+    assert "Preview execution" not in buttons
+    assert "Submit sequence" not in buttons
+    scheduler.client.proc.request_new.assert_not_called()