new_gui: pausable pipeline + run-plan stage toggles

- AppState: protocol (run-plan {center,raster,xrf,collect}), run_status
  (idle/running/paused/breakpoint), advance_past
- pipeline_panel: RunPlanRow chips (which stages run) + transport status uses
  run_status (Running/Paused/Stopped at breakpoint)
- manual_view: pausable auto-run runner — Play executes the lit stages in order,
  advances on raster/standard completion, busy True->False edge, or Skip; pauses
  before flagged breakpoints; Pause holds after current stage; End resets
- main_window: standard_scan_completed -> manual.on_collect_completed
- automation already pauses between samples (per-step within a sample is bounded
  by the backend automated_scan)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/aare/gui/new_gui/README.md

@@ -100,6 +100,13 @@ Backend contract is documented in the `new-gui-backend-paths` memory.
 - **Preconditions**: data-collection actions are gated by a ring/shutter/hutch
   warning (1-hour snooze); mounting is hard-blocked by the hutch PSS;
   `sample_missing` shows a modal (suppressed during automation).
+- **Pausable pipeline** (mounted): a **RUN PLAN** chip row toggles which stages
+  ▶ Play executes (Center/Raster/XRF/Collect). Play runs the lit stages in order,
+  advancing as each completes (`raster_scan_completed`/`standard_scan_completed`,
+  else a `status.busy` True→False edge, else ⏭ Skip) and **pausing before any
+  stage flagged with a ⚑ breakpoint**; ▶ resumes, ❚❚ pauses after the current
+  stage, ⏹ ends. (Automation pauses between samples; per-stage control inside a
+  sample is bounded by the backend's `automated_scan`.)
 - **Bookmarks** (mounted): ＋ Bookmark captures the current smargon position
   (shown as numbered markers on the camera, projected via `smargon_to_picture`);
   each has an **editable label**; **Go** returns there (`move_smargon` +

src/aare/gui/new_gui/main_window.py

@@ -325,6 +325,8 @@ class MainWindow(QWidget):
         cc.trace_clicked.connect(self._open_trace)
         if hasattr(self.daq, "raster_scan_completed"):
             self.daq.raster_scan_completed.connect(self.manual.on_raster_completed)
+        if hasattr(self.daq, "standard_scan_completed"):
+            self.daq.standard_scan_completed.connect(self.manual.on_collect_completed)
 
         # XRF spectrum view <-> fluorimeter
         spec = self.manual.spectrum

src/aare/gui/new_gui/manual_view.py

@@ -54,6 +54,13 @@ class ManualView(QWidget):
         self._grids_dialog = None
         self._bookmarks: list = []   # [{coord, omega, label}]
         self._bookmark_loaded_key = None
+        # pausable auto-run state
+        self._run_active = False
+        self._run_list: list[str] = []
+        self._run_idx = -1
+        self._cleared_bps: set[str] = set()
+        self._await_stage: str | None = None
+        self._seen_busy = False
 
         root = QHBoxLayout(self)
         root.setContentsMargins(0, 0, 0, 0)
@@ -158,14 +165,17 @@ class ManualView(QWidget):
         self.camera.grid_drawn.connect(self._on_grid_drawn)
         self.camera.clicked_point.connect(self._on_camera_click)
 
-        # transport
+        # transport (drives the pausable auto-run)
         t = self.pipeline.transport
-        t.play.connect(s.play)
-        t.pause.connect(s.pause)
-        t.skip.connect(s.skip)
-        t.end.connect(s.end)
+        t.play.connect(self._on_play)
+        t.pause.connect(self._on_pause)
+        t.skip.connect(self._on_skip)
+        t.end.connect(self._on_end)
         t.unmount.connect(s.unmount)
 
+        # run plan (which stages the auto-run executes)
+        self.pipeline.run_plan.toggled.connect(s.toggle_protocol)
+
         # bookmarks
         bm = self.pipeline.bookmarks
         bm.bookmark_add.connect(self._on_bookmark_add)
@@ -247,6 +257,111 @@ class ManualView(QWidget):
         )
         self.rotation_scan_requested.emit(req)
 
+    # ------------------------------------------------- pausable auto-run
+    # ▶ runs the enabled run-plan stages in order, advancing as each completes
+    # (raster/standard completion signals, else a busy True→False edge, else
+    # ⏭ Skip) and pausing before any stage flagged with a ⚑ breakpoint.
+    _STAGE_FIRE = {"center", "raster", "xrf", "collect"}
+
+    def _on_play(self) -> None:
+        s = self._state
+        if s.mount_phase != "mounted":
+            return
+        if not self._run_active:
+            self._run_list = s.enabled_run_stages()
+            if not self._run_list:
+                self.status_message.emit(
+                    "Run plan is empty — switch on some stages first.")
+                return
+            self._run_active = True
+            self._cleared_bps = set()
+            self._run_idx = -1
+            self._advance_run()
+        elif s.run_status == "breakpoint":
+            stage = self._run_list[self._run_idx]
+            self._cleared_bps.add(stage)
+            self._fire_stage(stage)
+        elif s.run_status == "paused":
+            if self._await_stage is not None:
+                s.set_run_status("running")     # a stage is still in flight
+            else:
+                self._advance_run()
+
+    def _on_pause(self) -> None:
+        if not self._run_active:
+            self._state.set_run_status("idle")
+            return
+        self._state.set_run_status("paused")
+
+    def _on_skip(self) -> None:
+        if not self._run_active:
+            self._state.skip()
+            return
+        self._await_stage = None
+        self._advance_run()
+
+    def _on_end(self) -> None:
+        self._run_active = False
+        self._await_stage = None
+        self._run_idx = -1
+        self._cleared_bps = set()
+        self._state.set_run_status("idle")
+        self._state.end()
+
+    def _advance_run(self) -> None:
+        s = self._state
+        self._await_stage = None
+        self._run_idx += 1
+        if self._run_idx >= len(self._run_list):
+            self._run_active = False
+            s.set_run_status("idle")
+            self.status_message.emit("Run plan complete.")
+            return
+        stage = self._run_list[self._run_idx]
+        s.pick_stage(stage)
+        if s.has_breakpoint(stage) and stage not in self._cleared_bps:
+            s.set_run_status("breakpoint")
+            return
+        self._fire_stage(stage)
+
+    def _fire_stage(self, stage: str) -> None:
+        s = self._state
+        if stage == "raster" and self._grid_rect_img is None:
+            self.status_message.emit(
+                "Raster is in the run plan but no grid is drawn — skipping it.")
+            self._advance_run()
+            return
+        s.set_run_status("running")
+        self._await_stage = stage
+        self._seen_busy = False
+        {"center": lambda: self._on_run_center("loop"),
+         "raster": self._on_run_raster,
+         "xrf": self._on_run_xrf,
+         "collect": self._on_run_collect}[stage]()
+
+    def _stage_done(self, stage: str) -> None:
+        if not self._run_active or self._await_stage != stage:
+            return
+        self._await_stage = None
+        # mark the stage done on the tracker
+        self._state.advance_past(stage)
+        if self._state.run_status == "paused":
+            return                              # honour a pause requested mid-stage
+        self._advance_run()
+
+    def on_collect_completed(self, *_a) -> None:
+        self._stage_done("collect")
+
+    def _check_busy_edge(self, s) -> None:
+        """Generic stage completion: when the op we fired clears ``busy``."""
+        if not self._run_active or self._await_stage is None:
+            return
+        busy = bool(getattr(s, "busy", False))
+        if busy:
+            self._seen_busy = True
+        elif self._seen_busy:
+            self._stage_done(self._await_stage)
+
     def _on_grid_drawn(self, x: float, y: float, w: float, h: float) -> None:
         self._grid_rect_img = (x, y, w, h)
         self.camera.set_heatmap(None)   # clear any prior result heatmap
@@ -263,6 +378,7 @@ class ManualView(QWidget):
         self._raster_grids = list(getattr(completed, "r", []) or [])
         self.pipeline.settings.set_raster_results_count(len(self._raster_grids))
         self._render_heatmap()
+        self._stage_done("raster")
 
     def _render_heatmap(self) -> None:
         if not self._raster_grids:
@@ -472,6 +588,7 @@ class ManualView(QWidget):
 
     def _refresh_pipe(self) -> None:
         self.pipeline.update_from_state(self._state)
+        self.pipeline.run_plan.set_protocol(self._state.protocol)
         # enable grid-draw affordance only while the raster stage is selected
         if self._state.stage != "raster":
             self.pipeline.settings.set_draw_active(False)
@@ -496,6 +613,7 @@ class ManualView(QWidget):
 
     def update_daq_status(self, s) -> None:
         self._last_status = s
+        self._check_busy_edge(s)
         self.motors.update_daq_status(s)
         self.camera_controls.update_daq_status(s)
         if self._bookmarks:

src/aare/gui/new_gui/state.py

@@ -111,9 +111,39 @@ class AppState(QObject):
         self._breaks: dict[str, bool] = {}
         self._running = False
         self._seq_idx = 0
+        self._protocol = default_protocol()      # manual run-plan (which stages run)
+        self._run_status = "idle"                # idle | running | paused | breakpoint
         self._queue: list[QueueItem] = []
         self._defaults = default_protocol()
 
+    # ----------------------------------------------------- run-plan (protocol)
+    @property
+    def protocol(self) -> dict[str, bool]:
+        return self._protocol
+
+    def is_enabled(self, key: str) -> bool:
+        return bool(self._protocol.get(key, True))
+
+    def toggle_protocol(self, key: str) -> None:
+        if key in PROTOCOL_KEYS:
+            self._protocol[key] = not self._protocol.get(key, False)
+            self.pipe_changed.emit()
+
+    def enabled_run_stages(self) -> list[str]:
+        """Interactive stages selected for the auto-run, in pipeline order."""
+        return [k for k in PROTOCOL_KEYS if self._protocol.get(k)]
+
+    # ----------------------------------------------------- run status
+    @property
+    def run_status(self) -> str:
+        return self._run_status
+
+    def set_run_status(self, status: str) -> None:
+        if status != self._run_status:
+            self._run_status = status
+            self._running = status == "running"
+            self.pipe_changed.emit()
+
     # ------------------------------------------------------------------ mode
     @property
     def mode(self) -> str:
@@ -204,6 +234,11 @@ class AppState(QObject):
         self._breaks[key] = not self._breaks.get(key)
         self.pipe_changed.emit()
 
+    def advance_past(self, key: str) -> None:
+        """Mark ``key`` complete on the tracker (sequence moves just past it)."""
+        self._seq_idx = max(self._seq_idx, self._stage_index(key) + 1)
+        self.pipe_changed.emit()
+
     def _stage_index(self, key: str) -> int:
         for i, s in enumerate(STAGES):
             if s.key == key:

src/aare/gui/new_gui/widgets/pipeline_panel.py

@@ -28,6 +28,7 @@ from PySide6.QtWidgets import (
 
 from aare.gui.new_gui.theme import FONT_MONO_FALLBACK, Palette
 from aare.gui.new_gui.widgets.common import (
+    Chip,
     micro_label,
     mono_label,
     vline,
@@ -35,6 +36,42 @@ from aare.gui.new_gui.widgets.common import (
 from aare.gui.new_gui.widgets.heatmap import METRICS
 from aare.gui.new_gui.widgets.pipeline import PipelineTracker
 
+_RUN_PLAN_STAGES = (("center", "Center"), ("raster", "Raster"),
+                    ("xrf", "XRF"), ("collect", "Collect"))
+
+
+class RunPlanRow(QWidget):
+    """Toggle which stages ▶ Play runs. ⚑ on a node pauses before that stage."""
+
+    toggled = Signal(str)
+
+    def __init__(self, palette: Palette, parent=None):
+        super().__init__(parent)
+        lay = QHBoxLayout(self)
+        lay.setContentsMargins(0, 0, 0, 0)
+        lay.setSpacing(6)
+        lay.addWidget(micro_label("RUN PLAN"))
+        self._chips: dict[str, Chip] = {}
+        for key, label in _RUN_PLAN_STAGES:
+            c = Chip(label, palette, checked=False, mono=False)
+            c.clicked.connect(lambda _=False, k=key: self.toggled.emit(k))
+            self._chips[key] = c
+            lay.addWidget(c)
+        hint = QLabel("▶ runs the lit stages · tap a stage's ⚑ to pause before it")
+        hint.setStyleSheet(f"font-size:10.5px; color:{palette.text_faint};")
+        lay.addSpacing(6)
+        lay.addWidget(hint)
+        lay.addStretch(1)
+
+    def set_protocol(self, protocol: dict) -> None:
+        for key, chip in self._chips.items():
+            on = bool(protocol.get(key))
+            if chip.isChecked() != on:
+                chip.blockSignals(True)
+                chip.setChecked(on)
+                chip.blockSignals(False)
+                chip._restyle()
+
 
 def _primary_button(text: str, palette: Palette) -> QPushButton:
     b = QPushButton(text)
@@ -583,25 +620,26 @@ class TransportRow(QWidget):
         )
 
     def update_from_state(self, state) -> None:
-        running = state.running
-        self._style_tbtn(self._play, active=running)
-        self._style_tbtn(self._pause, active=not running)
+        status = getattr(state, "run_status", "running" if state.running else "idle")
+        self._style_tbtn(self._play, active=status == "running")
+        self._style_tbtn(self._pause, active=status in ("paused", "breakpoint"))
         self._next.setText(state.next_stage().action_label)
-        if state.stopped_at_breakpoint():
+        if status == "breakpoint":
             self._status.setText("⏸ Stopped at breakpoint")
             self._status.setStyleSheet(
-                f"font-size:11.5px; color:{self._p.breakpoint}; font-weight:600;"
-            )
-        elif running:
+                f"font-size:11.5px; color:{self._p.breakpoint}; font-weight:600;")
+        elif status == "running":
             self._status.setText("● Running")
             self._status.setStyleSheet(
-                f"font-size:11.5px; color:{self._p.success}; font-weight:600;"
-            )
+                f"font-size:11.5px; color:{self._p.success}; font-weight:600;")
+        elif status == "paused":
+            self._status.setText("⏸ Paused")
+            self._status.setStyleSheet(
+                f"font-size:11.5px; color:{self._p.breakpoint}; font-weight:600;")
         else:
             self._status.setText("Idle")
             self._status.setStyleSheet(
-                f"font-size:11.5px; color:{self._p.text_faint};"
-            )
+                f"font-size:11.5px; color:{self._p.text_faint};")
 
 
 # --------------------------------------------------------------------------
@@ -659,6 +697,10 @@ class PipelinePanel(QWidget):
         self.settings = ContextualSettings(palette, defaults)
         lay.addWidget(self.settings)
 
+        # run plan (which stages ▶ Play runs)
+        self.run_plan = RunPlanRow(palette)
+        lay.addWidget(self.run_plan)
+
         # transport
         self.transport = TransportRow(palette)
         lay.addWidget(self.transport)
@@ -674,6 +716,7 @@ class PipelinePanel(QWidget):
         self._mounted = mounted
         self._badge.setVisible(mounted)
         self.bookmarks.setVisible(mounted)
+        self.run_plan.setVisible(mounted)
         self.transport.setVisible(mounted)
         self.settings.set_run_enabled(mounted)
         self.tracker.set_preview(not mounted)