fix(connector): isolate worker task exceptions and remove bare excepts

bec_widgets/utils/bec_connector.py

@@ -51,11 +51,16 @@ class ConnectionConfig(BaseModel):
 class WorkerSignals(QObject):
     progress = Signal(dict)
     completed = Signal()
+    failed = Signal(str)
 
 
 class Worker(QRunnable):
     """
     Worker class to run a function in a separate thread.
+
+    On success, ``signals.completed`` is emitted. If the function raises,
+    the exception is logged and ``signals.failed`` is emitted with the
+    formatted traceback instead; it never propagates into the thread pool.
     """
 
     def __init__(self, func, *args, **kwargs):
@@ -69,8 +74,29 @@ class Worker(QRunnable):
         """
         Run the specified function in the thread.
         """
-        self.func(*self.args, **self.kwargs)
-        self.signals.completed.emit()
+        try:
+            self.func(*self.args, **self.kwargs)
+        except Exception:
+            error_msg = traceback.format_exc()
+            logger.error(
+                f"Worker task {getattr(self.func, '__qualname__', self.func)} failed:\n{error_msg}"
+            )
+            self._emit_outcome(self.signals.failed, error_msg)
+        else:
+            self._emit_outcome(self.signals.completed)
+
+    @staticmethod
+    def _emit_outcome(signal, *args):
+        """Emit the outcome signal, tolerating a signal source deleted mid-task.
+
+        During application shutdown the Qt side of ``WorkerSignals`` can be destroyed
+        while the task is still running; the late emit must not escape ``run`` as an
+        unhandled exception in the thread pool.
+        """
+        try:
+            signal.emit(*args)
+        except RuntimeError:
+            logger.debug("Worker outcome signal emitted after its source was deleted")
 
 
 class BECConnector:
@@ -224,8 +250,9 @@ class BECConnector:
                 return None
             connector_parent = self._get_rpc_parent_ancestor()
             return connector_parent.gui_id if connector_parent else None
-        except:
-            logger.error(f"Error getting parent_id for {self.__class__.__name__}")
+        except Exception as e:
+            logger.error(f"Error getting parent_id for {self.__class__.__name__}: {e}")
+            return None
 
     def _get_rpc_parent_ancestor(self) -> BECConnector | None:
         """
@@ -375,8 +402,16 @@ class BECConnector:
             worker.signals.completed.connect(on_complete)
         # Keep a reference to the worker so it is not garbage collected.
         self._workers.append(worker)
-        # When the worker is done, remove it from our list.
-        worker.signals.completed.connect(lambda: self._workers.remove(worker))
+
+        # When the worker is done (success or failure), remove it from our list.
+        def _discard_worker(*_):
+            try:
+                self._workers.remove(worker)
+            except ValueError:
+                pass
+
+        worker.signals.completed.connect(_discard_worker)
+        worker.signals.failed.connect(_discard_worker)
         self._thread_pool.start(worker)
         return worker
 

bec_widgets/widgets/services/bec_queue/bec_queue.py

@@ -4,6 +4,7 @@ import json
 
 from bec_lib import messages
 from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
 from bec_qthemes import material_icon
 from qtpy.QtCore import Property, Qt, Signal, Slot
 from qtpy.QtGui import QColor
@@ -20,6 +21,8 @@ from bec_widgets.widgets.control.buttons.button_reset.button_reset import ResetB
 from bec_widgets.widgets.control.buttons.button_resume.button_resume import ResumeButton
 from bec_widgets.widgets.control.buttons.stop_button.stop_button import StopButton
 
+logger = bec_logger.logger
+
 
 class BECQueue(BECWidget, CompactPopupWidget):
     """
@@ -243,8 +246,8 @@ class BECQueue(BECWidget, CompactPopupWidget):
             try:
                 color = self.status_colors.get(content, "black")  # Default to black if not found
                 item.setForeground(QColor(color))
-            except:
-                return item
+            except Exception as e:
+                logger.warning(f"Could not apply status color for queue item '{content}': {e}")
         return item
 
     def set_row(

tests/unit_tests/conftest.py

@@ -390,3 +390,33 @@ def scan_history_factory(tmpdir):
         return create_history_file(file_path, data, metadata)
 
     return _factory
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _register_worker_thread_dummies():
+    """Qt thread-pool threads register a permanent ``threading._DummyThread`` the first
+    time Python inspects them (e.g. a loguru call inside a background Worker). Saturate
+    the global pool once up front - with thread expiry disabled so the threads persist -
+    so bec_lib's threads_check fixture never sees these registrations appear mid-test."""
+    import threading
+
+    from qtpy.QtCore import QRunnable, QThreadPool
+
+    pool = QThreadPool.globalInstance()
+    pool.setExpiryTimeout(-1)
+    barrier = threading.Barrier(pool.maxThreadCount() + 1, timeout=10)
+
+    class _Warmup(QRunnable):
+        def run(self):
+            threading.current_thread()
+            try:
+                barrier.wait()
+            except threading.BrokenBarrierError:
+                pass
+
+    for _ in range(pool.maxThreadCount()):
+        pool.start(_Warmup())
+    try:
+        barrier.wait()
+    except threading.BrokenBarrierError:
+        pass

tests/unit_tests/test_bec_connector.py

@@ -243,3 +243,37 @@ def test_bec_connector_terminate_registered_once_qapp_exists(qtbot):
         handler = BECConnector.EXIT_HANDLERS.pop(fresh_client, None)
         if handler is not None:
             QApplication.instance().aboutToQuit.disconnect(handler)
+
+
+def test_bec_connector_submit_task_failure_removes_worker(bec_connector, qtbot):
+    """A task that raises must not leak the worker
+    reference, must emit failed, and must not call on_complete."""
+    import threading
+
+    failures = []
+    completed = []
+    # the worker starts before submit_task returns, so the raise is gated until the
+    # failed connection below exists - otherwise this test races its own setup
+    connected = threading.Event()
+
+    def boom():
+        connected.wait(timeout=5)
+        raise RuntimeError("task failed on purpose")
+
+    worker = bec_connector.submit_task(boom, on_complete=lambda: completed.append(True))
+    worker.signals.failed.connect(lambda msg: failures.append(msg))
+    connected.set()
+
+    qtbot.waitUntil(lambda: worker not in bec_connector._workers, timeout=5000)
+    qtbot.waitUntil(lambda: len(failures) == 1, timeout=5000)
+    assert completed == []
+    assert "task failed on purpose" in failures[0]
+
+
+def test_bec_connector_parent_id_returns_none_on_error(bec_connector):
+    """parent_id must swallow only Exception and
+    return None explicitly."""
+    with mock.patch.object(
+        bec_connector, "_get_rpc_parent_ancestor", side_effect=ValueError("broken hierarchy")
+    ):
+        assert bec_connector.parent_id is None