wip

debye_bec/bec_widgets/widgets/scheduler/scheduler.py

@@ -843,10 +843,42 @@ class Scheduler(BECWidget, QWidget):
         """Must be called while holding `self._lock`; see `schedule_logic.pick_next_runnable`."""
         return pick_next_runnable(self.schedule.items, repeat_aborted_item)
 
+    @SafeSlot()
+    def _on_run_finished(self):
+        self._refresh_ui()
+
+    @SafeSlot(str)
+    def _on_run_failed(self, error: str):
+        logger.error(f"Schedule execution task failed unexpectedly: {error}")
+        with self._lock:
+            self.schedule.is_running = False
+            self._persist_locked()
+        self._refresh_ui()
+
+    def _emit_schedule_changed(self):
+        if self._closing.is_set():
+            return
+
+        self.schedule_changed.emit()
+
+    def closeEvent(self, event):
+        """Ensure cleanup is invoked when the widget is closed via the window system."""
+        self.cleanup()
+        super().closeEvent(event)
+
+    def cleanup(self):
+        """Stop background execution loops without cancelling server-side scans."""
+        self._closing.set()
+
+        # Interrupt guard wait loops if any thread is blocked in self._guard.wait_until_clear()
+        self._guard.cleanup()
+
+        super().cleanup()
+
     def _run_all(self, repeat_aborted_item):
         logger.info("_run_all was called")
         namespace = {"scans": self.scans, "dev": self.dev}
-        while not self._abort_requested:
+        while not self._abort_requested and not self._closing.is_set():
             with self._lock:
                 next_item = self._pick_next_runnable_locked(repeat_aborted_item)
             if next_item is None or next_item == "stop":
@@ -865,12 +897,12 @@ class Scheduler(BECWidget, QWidget):
 
     def _execute_item(self, item: ScheduleItem, namespace: dict):
         if item.kind == "scan":
-            # Block here (item stays PENDING - nothing "started" yet) until
-            # the guard is clear or the operator aborts. See "Guard
-            # interaction" in the module docstring.
-            cleared = self._guard.wait_until_clear(should_abort=lambda: self._abort_requested)
+            # Block here until the guard is clear, operator aborts, or widget closes
+            cleared = self._guard.wait_until_clear(
+                should_abort=lambda: self._abort_requested or self._closing.is_set()
+            )
             self._emit_schedule_changed()
-            if not cleared:
+            if not cleared or self._closing.is_set():
                 return
 
         with self._lock:
@@ -890,12 +922,15 @@ class Scheduler(BECWidget, QWidget):
             request = getattr(report, "request", None)
             with self._lock:
                 item.request_id = getattr(request, "requestID", None)
-                # persist the request_id right away, before the
-                # (potentially long) wait below, so a reconnect can find
-                # it even if this process disappears mid-scan.
                 self._persist_locked()
 
-            report.wait()
+            # Wait for completion without blocking the thread indefinitely
+            self._wait_for_report(report)
+
+            if self._closing.is_set():
+                # The widget was closed mid-item; exit without changing status in Redis.
+                # Reconciliation logic will handle state on reopen.
+                return
 
             scan = getattr(report, "scan", None)
             with self._lock:
@@ -907,12 +942,10 @@ class Scheduler(BECWidget, QWidget):
                     else ScheduleItemStatus.COMPLETED
                 )
         except Exception:  # pylint: disable=broad-except
+            if self._closing.is_set():
+                return
             with self._lock:
                 if self._guard_interrupt_requested:
-                    # Interrupted by the auto-pause guard, not a real
-                    # failure/operator abort - reset to PENDING so the
-                    # normal execution loop retries it once the guard
-                    # clears, instead of leaving it in a terminal state.
                     item.status = ScheduleItemStatus.PENDING
                     item.request_id = None
                     item.scan_id = None
@@ -927,14 +960,17 @@ class Scheduler(BECWidget, QWidget):
                     item.error = traceback.format_exc()
                     logger.error(f"Schedule item failed: {item.command}\n{item.error}")
         finally:
-            with self._lock:
-                if item.status != ScheduleItemStatus.PENDING:
-                    item.finished_at = time.time()
-                self._persist_locked()
-                final_status = item.status
-            self._current_report = None
-            self._current_item_kind = None
-            self._emit_schedule_changed()
+            if not self._closing.is_set():
+                with self._lock:
+                    if item.status != ScheduleItemStatus.PENDING:
+                        item.finished_at = time.time()
+                    self._persist_locked()
+                    final_status = item.status
+                self._current_report = None
+                self._current_item_kind = None
+                self._emit_schedule_changed()
+            else:
+                final_status = None
 
         if final_status in (ScheduleItemStatus.COMPLETED, ScheduleItemStatus.FAILED):
             notify_item_finished(
@@ -947,9 +983,22 @@ class Scheduler(BECWidget, QWidget):
                 error=item.error,
             )
 
+    def _wait_for_report(self, report, timeout: float = 0.5):
+        """Poll report status to allow thread exit if the widget closes."""
+        while not self._closing.is_set() and not self._abort_requested:
+            # Check if scan report has finished via its internal event or status
+            if (
+                getattr(report, "status", None) == "completed"
+                or getattr(report, "event", None)
+                and report.event.is_set()
+            ):
+                break
+            # Non-blocking poll interval
+            time.sleep(timeout)
+
     def _await_running_item(self, item: ScheduleItem):
         queue_storage = self.client.queue.queue_storage
-        while not self._abort_requested:
+        while not self._abort_requested and not self._closing.is_set():
             queue_item = queue_storage.find_queue_item_by_requestID(item.request_id)
             if queue_item is None or not self._is_queue_item_active(queue_item):
                 with self._lock:
@@ -959,29 +1008,3 @@ class Scheduler(BECWidget, QWidget):
                 self._emit_schedule_changed()
                 return
             time.sleep(0.5)
-
-    @SafeSlot()
-    def _on_run_finished(self):
-        self._refresh_ui()
-
-    @SafeSlot(str)
-    def _on_run_failed(self, error: str):
-        logger.error(f"Schedule execution task failed unexpectedly: {error}")
-        with self._lock:
-            self.schedule.is_running = False
-            self._persist_locked()
-        self._refresh_ui()
-
-    def _emit_schedule_changed(self):
-        if self._closing.is_set():
-            return
-
-        self.schedule_changed.emit()
-
-    def cleanup(self):
-        # The schedule itself must continue running after the widget is closed.
-        # We only tell the worker that the UI/Qt object is no longer available.
-        self._closing.set()
-
-        self._guard.cleanup()
-        super().cleanup()