fix(connector): connect on_failed callbacks before the worker starts

bec_widgets/utils/bec_connector.py

@@ -369,7 +369,9 @@ class BECConnector:
             self.rpc_register.mark_broadcast_pending()
             self.rpc_register.broadcast()
 
-    def submit_task(self, fn, *args, on_complete: SafeSlot = None, **kwargs) -> Worker:
+    def submit_task(
+        self, fn, *args, on_complete: SafeSlot = None, on_failed: SafeSlot = None, **kwargs
+    ) -> Worker:
         """
         Submit a task to run in a separate thread. The task will run the specified
         function with the provided arguments and emit the completed signal when done.
@@ -380,7 +382,11 @@ class BECConnector:
         Args:
             fn: Function to run in a separate thread.
             *args: Arguments for the function.
-            on_complete: Slot to run when the task is complete.
+            on_complete: Slot to run when the task completes successfully.
+            on_failed: Slot to run when the task raises; receives the formatted traceback
+                string. Pass it here rather than connecting to ``worker.signals.failed``
+                after this method returns: the worker starts before this method returns,
+                so a late connection can miss the emission of a fast-failing task.
             **kwargs: Keyword arguments for the function.
 
         Returns:
@@ -400,6 +406,8 @@ class BECConnector:
         worker = Worker(fn, *args, **kwargs)
         if on_complete:
             worker.signals.completed.connect(on_complete)
+        if on_failed:
+            worker.signals.failed.connect(on_failed)
         # Keep a reference to the worker so it is not garbage collected.
         self._workers.append(worker)
 

tests/unit_tests/test_bec_connector.py

@@ -300,3 +300,41 @@ def test_bec_connector_worker_completion_does_not_retain_owner(qtbot, mocked_cli
         gc.collect()
 
     assert ref() is None, "connector kept alive by worker completion closure"
+
+
+def test_bec_connector_on_failed_never_misses_fast_failures(bec_connector, qtbot):
+    """``on_failed`` passed to submit_task is connected before the worker starts, so
+    even a task that raises immediately cannot emit ``failed`` before the connection
+    exists. Connecting to ``worker.signals.failed`` after submit_task returns cannot
+    give this guarantee."""
+    failures = []
+
+    def boom():
+        raise RuntimeError("instant failure")
+
+    for _ in range(20):
+        bec_connector.submit_task(boom, on_failed=lambda msg: failures.append(msg))
+
+    qtbot.waitUntil(lambda: len(failures) == 20, timeout=5000)
+    qtbot.waitUntil(lambda: not bec_connector._workers, timeout=5000)
+    assert all("instant failure" in msg for msg in failures)
+
+
+def test_bec_connector_worker_outcome_survives_deleted_signal_source(bec_connector, qtbot, capfd):
+    """At application shutdown the WorkerSignals C++ object can die before a late worker
+    finishes; the outcome emit must not escape Worker.run into the thread pool."""
+    import threading
+
+    import shiboken6
+
+    gate = threading.Event()
+    worker = bec_connector.submit_task(lambda: gate.wait(timeout=5))
+    shiboken6.delete(worker.signals)
+    gate.set()
+
+    qtbot.wait(300)  # let the worker finish and attempt both emits
+    stderr = capfd.readouterr().err
+    assert "Error calling Python override of QRunnable::run()" not in stderr
+    # the discard connection died with the signal source; drop the worker manually
+    if worker in bec_connector._workers:
+        bec_connector._workers.remove(worker)