Add killed state handling to SweepQueue and internal sweep flag

Introduces a '_last_killed' flag to SweepQueue to track when the queue is stopped via kill() or kill_all(), and updates status reporting to include a 'killed' effective state. Internal ramp and inner sweeps are now marked with a '_internal_sweep' attribute to prevent sweep collision checks. Unit tests are added to verify the new killed state behavior.

Author: caidish

src/measureit/sweep/base_sweep.py

@@ -566,7 +566,11 @@ def start(self, persist_data=None, ramp_to_start=False):
             Optional argument which gradually ramps each parameter to the starting
             point of its sweep. Default is true for Sweep1D and Sweep2D.
         """
-        if not self.progressState.is_queued and _has_other_active_sweep(self):
+        if (
+            not self.progressState.is_queued
+            and not getattr(self, "_internal_sweep", False)
+            and _has_other_active_sweep(self)
+        ):
             raise RuntimeError(
                 "Another sweep is already running. Stop or kill it before starting a new sweep, use start_force() to kill others, or use SweepQueue to stack sweeps."
             )
@@ -645,7 +649,11 @@ def start(self, persist_data=None, ramp_to_start=False):
     def resume(self):
         """Restarts the sweep after it has been paused."""
         if self.progressState.state == SweepState.PAUSED:
-            if not self.progressState.is_queued and _has_other_active_sweep(self):
+            if (
+                not self.progressState.is_queued
+                and not getattr(self, "_internal_sweep", False)
+                and _has_other_active_sweep(self)
+            ):
                 raise RuntimeError(
                     "Another sweep is already running. Stop or kill it before resuming this sweep, use start_force() to kill others, or use SweepQueue to stack sweeps."
                 )

src/measureit/sweep/simul_sweep.py

@@ -406,6 +406,7 @@ def ramp_to(self, vals_dict, start_on_finish=False, persist=None, multiplier=1):
             suppress_output=self.suppress_output,
         )
         self.ramp_sweep.parent = self
+        self.ramp_sweep._internal_sweep = True
         # Only follow parameters that are not being ramped to avoid circular dependencies
         follow_params = [p for p in self._params if p not in ramp_params_dict.keys()]
         if follow_params:

src/measureit/sweep/sweep1d.py

@@ -507,6 +507,7 @@ def ramp_to(self, value, start_on_finish=False, persist=None, multiplier=1):
             plot_data=self.plot_data,
         )
         self.ramp_sweep.parent = self
+        self.ramp_sweep._internal_sweep = True
         self.ramp_sweep.follow_param(self._params)
 
         self.progressState.state = SweepState.RAMPING

src/measureit/sweep/sweep1d_listening.py

@@ -402,6 +402,7 @@ def ramp_to(self, value, start_on_finish=False, persist=None, multiplier=1):
             plot_data=self.plot_data,
         )
         self.ramp_sweep.parent = self
+        self.ramp_sweep._internal_sweep = True
 
         self.progressState.state = SweepState.RAMPING
         self.ramp_sweep.start(ramp_to_start=False)

src/measureit/sweep/sweep2d.py

@@ -216,6 +216,7 @@ def __init__(
         )
         # Set parent reference so UI actions (e.g., ESC) can stop both inner and outer sweeps
         self.in_sweep.parent = self
+        self.in_sweep._internal_sweep = True
         # Ensure the inner sweep writes metadata for the composite (outer) sweep
         self.in_sweep.metadata_provider = self
         # We set our outer sweep parameter as a follow param for the inner sweep, so that
@@ -674,6 +675,7 @@ def ramp_to(self, value, start_on_finish=False, multiplier=1):
             plot_data=True,
         )
         self.ramp_sweep.parent = self
+        self.ramp_sweep._internal_sweep = True
         for p in self._params:
             if p is not self.set_param:
                 self.ramp_sweep.follow_param(p)
@@ -707,6 +709,7 @@ def ramp_to_zero(self):
             complete_func=self.done_ramping,
         )
         self.ramp_sweep.parent = self
+        self.ramp_sweep._internal_sweep = True
         self.ramp_sweep.follow_param(self._params)
         self.progressState.state = SweepState.RAMPING
         self.outer_ramp = True

src/measureit/tools/sweep_queue.py

@@ -199,6 +199,8 @@ def __init__(self, inter_delay=1, post_db_delay=1.0, debug=False):
         self.previous_action = None
         # Queue-level error state (persists after sweep cleanup)
         self._last_error: Optional[QueueError] = None
+        # Queue-level killed flag (persists after sweep cleanup)
+        self._last_killed = False
 
     def _exec_in_kernel(self, fn):
         """Execute a callable synchronously.
@@ -511,6 +513,8 @@ def start(self, rts=True):
 
         self.log.info("Starting sweeps")
         self.current_action = self.queue.popleft()
+        # Clear kill flag once new work begins
+        self._last_killed = False
         if isinstance(self.current_action, BaseSweep):
             self.current_sweep = self.current_action
             # Ensure metadata shows this sweep was launched by SweepQueue
@@ -592,6 +596,7 @@ def resume(self):
     def kill(self):
         """Kills the current sweep. Use kill_all() to also clear the queue."""
         if self.current_sweep is not None:
+            self._last_killed = True
             # Clear current_sweep AND current_action before kill() to prevent
             # begin_next() from processing if kill() emits completion synchronously
             sweep_to_kill = self.current_sweep
@@ -609,6 +614,11 @@ def kill_all(self):
         Note: If a DatabaseEntry or callable is currently executing, it cannot be
         interrupted, but the queue will not continue after it completes.
         """
+        had_work = (
+            self.current_sweep is not None
+            or self.current_action is not None
+            or len(self.queue) > 0
+        )
         # Save reference to sweep before clearing state
         sweep_to_kill = self.current_sweep
 
@@ -625,6 +635,8 @@ def kill_all(self):
 
         # Clear error state for full reset
         self._last_error = None
+        # Mark queue as killed if there was work to stop
+        self._last_killed = had_work
 
         # Now safe to kill - even if completion fires, queue is empty
         if sweep_to_kill is not None:
@@ -650,7 +662,7 @@ def status(self):
 
         Returns a dictionary with:
         - effective_state: Overall queue state accounting for pending items
-          ("idle", "pending", "running", "paused", "error", "stopped")
+          ("idle", "pending", "running", "paused", "killed", "error", "stopped")
         - current_sweep_state: State name of the currently executing sweep (or None)
         - queue_length: Number of items waiting in the queue
         - current_sweep_type: Class name of current sweep (or None)
@@ -667,6 +679,7 @@ def status(self):
         - "running": Current sweep is actively running/ramping, or a DatabaseEntry/callable
           is executing
         - "paused": Current sweep is paused
+        - "killed": Queue stopped due to a kill() or kill_all() call. Call start() to resume.
         - "error": Current sweep is in error state (actively erroring)
         - "stopped": Queue stopped due to a previous error (check last_error for details).
                      Call clear_error() and start() to resume.
@@ -704,12 +717,16 @@ def status(self):
                 effective_state = "running"
             elif self._last_error is not None:
                 effective_state = "stopped"
+            elif self._last_killed:
+                effective_state = "killed"
             elif queue_length > 0:
                 effective_state = "pending"
             else:
                 effective_state = "idle"
         elif current_sweep_state == SweepState.ERROR:
             effective_state = "error"
+        elif current_sweep_state == SweepState.KILLED:
+            effective_state = "killed"
         elif current_sweep_state == SweepState.PAUSED:
             effective_state = "paused"
         elif current_sweep_state in (SweepState.RUNNING, SweepState.RAMPING):
@@ -724,7 +741,7 @@ def status(self):
             else:
                 effective_state = "idle"
         else:
-            # KILLED or unexpected states with a current_sweep still set
+            # Unexpected states with a current_sweep still set
             if queue_length > 0:
                 effective_state = "pending"
             else:
@@ -799,11 +816,12 @@ def begin_next(self):
         self._processing = True
         self._pending_begin_next = False
 
-        # Guard: if queue is in error state, don't process further
-        # This prevents re-entrancy from draining the queue after an error
-        if self._last_error is not None:
+        # Guard: if queue is in error or killed state, don't process further
+        # This prevents re-entrancy from draining the queue after a stop
+        if self._last_error is not None or self._last_killed:
             if self.debug:
-                self.log.debug("Queue in error state, not processing further")
+                state = "error" if self._last_error is not None else "killed"
+                self.log.debug("Queue in %s state, not processing further", state)
             self._processing = False
             return
 

tests/unit/test_sweep_queue.py

@@ -17,6 +17,7 @@
 os.environ.setdefault("MEASUREIT_FAKE_QT", "1")
 
 from measureit.tools import sweep_queue as sq
+from measureit.sweep.progress import SweepState
 
 
 class DummySignal:
@@ -54,7 +55,7 @@ def __init__(
         self.killed: int = 0
         self.resumed: int = 0
         self.metadata_provider = None
-        self.progressState = type("State", (), {"state": "READY"})()
+        self.progressState = type("State", (), {"state": SweepState.READY})()
         self.completed = DummySignal()
         self._complete_func: Optional[Callable[[], None]] = None
         self.begin = start_value
@@ -81,15 +82,15 @@ def set_complete_func(self, func: Callable[..., None], *args: Any, **kwargs: Any
 
     def start(self, persist_data: Any = None, ramp_to_start: bool = False) -> None:
         self.started = True
-        self.progressState.state = "RUNNING"
+        self.progressState.state = SweepState.RUNNING
 
     def kill(self) -> None:
         self.killed += 1
-        self.progressState.state = "KILLED"
+        self.progressState.state = SweepState.KILLED
 
     def resume(self) -> None:
         self.resumed += 1
-        self.progressState.state = "RUNNING"
+        self.progressState.state = SweepState.RUNNING
 
     def export_json(self, fn: Optional[str] = None) -> dict[str, Any]:
         return dict(self.export_payload)
@@ -98,7 +99,7 @@ def export_json(self, fn: Optional[str] = None) -> dict[str, Any]:
 
     def trigger_complete(self) -> None:
         """Simulate the sweep finishing and emitting its completed signal."""
-        self.progressState.state = "DONE"
+        self.progressState.state = SweepState.DONE
         self.completed.emit()
 
     def __repr__(self) -> str:  # pragma: no cover - debugging helper
@@ -235,3 +236,31 @@ def test_export_json_includes_queue_configuration(queue: sq.SweepQueue) -> None:
     assert data["inter_delay"] == queue.inter_delay
     assert isinstance(data["queue"], list)
     assert data["queue"][0]["attributes"]["name"] == "json-test"
+
+
+def test_status_reports_killed_after_kill(queue: sq.SweepQueue) -> None:
+    first = DummySweep1D("first")
+    second = DummySweep("second")
+
+    queue.append(first, second)
+    queue.start()
+
+    queue.kill()
+
+    status = queue.status()
+    assert status["effective_state"] == "killed"
+    assert status["queue_length"] == 1
+
+
+def test_start_clears_killed_state(queue: sq.SweepQueue) -> None:
+    first = DummySweep1D("first")
+    second = DummySweep("second")
+
+    queue.append(first, second)
+    queue.start()
+    queue.kill()
+
+    queue.start()
+
+    status = queue.status()
+    assert status["effective_state"] == "running"