diff --git a/scripts/ewoks_executor_demo.py b/scripts/ewoks_executor_demo.py
new file mode 100644
index 00000000..4ae5626e
--- /dev/null
+++ b/scripts/ewoks_executor_demo.py
@@ -0,0 +1,153 @@
+import logging
+import sys
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import ThreadPoolExecutor
+from functools import partial
+
+from AnyQt.QtWidgets import QApplication
+from AnyQt.QtWidgets import QGridLayout
+from AnyQt.QtWidgets import QHBoxLayout
+from AnyQt.QtWidgets import QLabel
+from AnyQt.QtWidgets import QListWidget
+from AnyQt.QtWidgets import QListWidgetItem
+from AnyQt.QtWidgets import QPushButton
+from AnyQt.QtWidgets import QTextEdit
+from AnyQt.QtWidgets import QVBoxLayout
+from AnyQt.QtWidgets import QWidget
+from ewokscore.tests.examples.tasks.sumtask import SumTask
+
+from ewoksorange.gui.concurrency.executor import EwoksExecutor
+from ewoksorange.gui.concurrency.executor import SubmitPolicy
+
+
+class Window(QWidget):
+    def __init__(self):
+        super().__init__()
+
+        self.setWindowTitle("EwoksExecutor Demo")
+
+        self.log = QTextEdit(readOnly=True)
+
+        self.pending = QListWidget()
+        self.pending_items = {}
+
+        self.executors = {
+            "T DROP": EwoksExecutor(
+                ThreadPoolExecutor(max_workers=1),
+                SubmitPolicy.DROP_IF_BUSY,
+            ),
+            "T QUEUE": EwoksExecutor(
+                ThreadPoolExecutor(max_workers=1),
+                SubmitPolicy.ALWAYS,
+            ),
+            "T PARALLEL": EwoksExecutor(
+                ThreadPoolExecutor(max_workers=4),
+                SubmitPolicy.ALWAYS,
+            ),
+            "P DROP": EwoksExecutor(
+                ProcessPoolExecutor(max_workers=1),
+                SubmitPolicy.DROP_IF_BUSY,
+            ),
+            "P QUEUE": EwoksExecutor(
+                ProcessPoolExecutor(max_workers=1),
+                SubmitPolicy.ALWAYS,
+            ),
+            "P PARALLEL": EwoksExecutor(
+                ProcessPoolExecutor(max_workers=4),
+                SubmitPolicy.ALWAYS,
+            ),
+        }
+
+        for name, executor in self.executors.items():
+            executor.submitted.connect(partial(self.on_submitted, name))
+            executor.ignored.connect(partial(self.on_ignored, name))
+            executor.succeeded.connect(partial(self.on_succeeded, name))
+            executor.failed.connect(partial(self.on_failed, name))
+
+        layout = QVBoxLayout(self)
+        grid = QGridLayout()
+
+        buttons = [
+            ("Submit x5 (Thread Drop)", "T DROP"),
+            ("Submit x5 (Process Drop)", "P DROP"),
+            ("Submit x5 (Thread Queue)", "T QUEUE"),
+            ("Submit x5 (Process Queue)", "P QUEUE"),
+            ("Submit x5 (Thread Parallel)", "T PARALLEL"),
+            ("Submit x5 (Process Parallel)", "P PARALLEL"),
+        ]
+
+        for i, (text, key) in enumerate(buttons):
+            button = QPushButton(text)
+            button.clicked.connect(partial(self.on_button_clicked, key))
+            grid.addWidget(button, i // 2, i % 2)
+
+        layout.addLayout(grid)
+
+        body = QHBoxLayout()
+
+        log_box = QVBoxLayout()
+        log_box.addWidget(QLabel("Logs"))
+        log_box.addWidget(self.log)
+        body.addLayout(log_box, 2)
+
+        pending_box = QVBoxLayout()
+        pending_box.addWidget(QLabel("Pending jobs"))
+        pending_box.addWidget(self.pending)
+        body.addLayout(pending_box, 1)
+
+        layout.addLayout(body)
+
+        self.counter = 1
+
+    def on_submitted(self, name, future):
+        if future is None:
+            self.log.append(f"[{name}] Submitted -> None")
+        else:
+            self.log.append(f"[{name}] Submitted -> {id(future)}")
+            item = QListWidgetItem(f"[{name}] {id(future)}")
+            self.pending.addItem(item)
+            self.pending_items[id(future)] = item
+
+    def on_ignored(self, name):
+        self.log.append(f"[{name}] ⚠ Ignored")
+
+    def on_succeeded(self, name, future, result):
+        values = {k: v.value for k, v in result.items()}
+        self.log.append(f"[{name}] ✅ {id(future)} -> {values}")
+        self.remove_pending(future)
+
+    def on_failed(self, name, future, exc):
+        self.log.append(f"[{name}] ❌ {id(future)} -> {exc}")
+        self.remove_pending(future)
+
+    def remove_pending(self, future):
+        item = self.pending_items.pop(id(future), None)
+        if item is not None:
+            self.pending.takeItem(self.pending.row(item))
+
+    def on_button_clicked(self, key, _checked=False):
+        self.submit_many(self.executors[key])
+
+    def submit_many(self, executor):
+        for _ in range(5):
+            executor.submit_task(
+                SumTask, inputs={"a": self.counter, "b": self.counter, "delay": 2}
+            )
+            self.counter += 1
+
+    def closeEvent(self, event):
+        for executor in self.executors.values():
+            executor.shutdown()
+        super().closeEvent(event)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+
+    app = QApplication(sys.argv)
+
+    window = Window()
+    window.resize(900, 500)
+    window.show()
+
+    sys.exit(app.exec())
diff --git a/src/ewoksorange/gui/concurrency/executor/EwoksExecutor.py b/src/ewoksorange/gui/concurrency/executor/EwoksExecutor.py
new file mode 100644
index 00000000..d9d1e5cb
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/EwoksExecutor.py
@@ -0,0 +1,214 @@
+import logging
+import multiprocessing
+import multiprocessing.managers
+import threading
+import weakref
+from concurrent.futures import Executor
+from concurrent.futures import Future
+from concurrent.futures import ProcessPoolExecutor
+from enum import Enum
+from enum import auto
+from threading import Event
+from threading import Lock
+from typing import Optional
+from typing import Type
+
+from AnyQt.QtCore import QObject
+from AnyQt.QtCore import Signal
+from ewokscore.task import Task
+
+from ._EwoksProcessHandle import EwoksProcessHandle as _EwoksProcessHandle
+from ._EwoksThreadHandle import EwoksThreadHandle as _EwoksThreadHandle
+from ._ProcessCallable import ProcessCallable as _ProcessCallable
+from .TaskFuture import TaskFuture
+
+_logger = logging.getLogger(__name__)
+
+
+class SubmitPolicy(Enum):
+    ALWAYS = auto()
+    DROP_IF_BUSY = auto()
+
+
+class EwoksExecutor(QObject):
+    """Qt-aware executor for ewoks tasks.
+
+    Wraps a concurrent.futures.Executor (thread or process) and emits Qt
+    signals on task lifecycle events.  Returns TaskFuture objects that support
+    both native future cancellation (cancel()) and ewoks task abort (abort()).
+
+    The `started` signal is always emitted in the main thread regardless of
+    the executor type.
+
+    Warning: the `ProcessPoolExecutor` implementation is a beta-version.
+    """
+
+    submitted = Signal(object)
+    """Emitted when a task is submitted. Argument is the TaskFuture."""
+    started = Signal(object)
+    """Emitted when a task starts executing. Argument is the TaskFuture."""
+    succeeded = Signal(object, object)
+    """Emitted when a task finishes successfully. Arguments are the TaskFuture and the result dict."""
+    failed = Signal(object, object)
+    """Emitted when a task raises an exception. Arguments are the TaskFuture and the exception."""
+    ignored = Signal()
+    """Emitted when a task submission is ignored due to the DROP_IF_BUSY policy."""
+    aborted = Signal(object)
+    """Emitted when a task was aborted. Argument is the TaskFuture."""
+    finished = Signal(object)
+    """Emitted when a task finishes (success, failure, or abort). Argument is the TaskFuture."""
+
+    def __init__(
+        self,
+        executor: Executor,
+        policy: SubmitPolicy = SubmitPolicy.ALWAYS,
+    ):
+        super().__init__()
+        self._executor = executor
+        self._policy = policy
+        self._running = 0
+        self._lock = Lock()
+        self._is_process = isinstance(executor, ProcessPoolExecutor)
+        # Manager is started lazily on first process submission and provides
+        # picklable proxy Queue / Event objects for IPC.
+        self._manager: Optional[multiprocessing.managers.SyncManager] = None
+
+    def submit_task(
+        self,
+        task_class: Type[Task],
+        **task_kwargs,
+    ) -> Optional[TaskFuture]:
+        """Submit an ewoks task for execution.
+
+        task_kwargs are forwarded directly to the task class constructor
+        (e.g. ``inputs={"a": 1}``).  Returns a TaskFuture, or None if the
+        policy dropped the submission.
+        """
+        with self._lock:
+            if self._policy is SubmitPolicy.DROP_IF_BUSY and self._running:
+                _logger.warning("Submission ignored: executor busy")
+                self.ignored.emit()
+                return None
+            self._running += 1
+
+        if self._is_process:
+            return self._submit_process(task_class, task_kwargs)
+        return self._submit_thread(task_class, task_kwargs)
+
+    def _submit_thread(self, task_class, task_kwargs) -> TaskFuture:
+        worker = _EwoksThreadHandle(task_class, **task_kwargs)
+        self_ref = weakref.ref(self)
+
+        # _ready gates the worker until task_future is assigned so that
+        # `started` is never emitted with a None reference.
+        _ready: Event = Event()
+        _holder: list = [None]
+
+        def _run():
+            _ready.wait()
+            self._emit_started(self_ref, _holder)
+            return worker()
+
+        raw_future = self._executor.submit(_run)
+        return self._finalize_submission(
+            raw_future, worker, self_ref, _holder, ready=_ready
+        )
+
+    def _submit_process(self, task_class, task_kwargs) -> TaskFuture:
+        manager = self._get_manager()
+        started_queue = manager.Queue()
+        abort_event = manager.Event()
+        aborted_event = manager.Event()
+
+        callable_obj = _ProcessCallable(
+            task_class, task_kwargs, started_queue, abort_event, aborted_event
+        )
+        worker = _EwoksProcessHandle(abort_event, aborted_event)
+        self_ref = weakref.ref(self)
+        _holder: list = [None]
+
+        # A daemon thread blocks on the queue and relays "started" to the
+        # Qt main thread once the subprocess signals it.
+        def _relay_started():
+            try:
+                msg = started_queue.get(timeout=300)
+                if msg == "started":
+                    self._emit_started(self_ref, _holder)
+            except Exception:
+                _logger.debug("started relay timed out or failed", exc_info=True)
+
+        threading.Thread(target=_relay_started, daemon=True).start()
+
+        raw_future = self._executor.submit(callable_obj)
+        return self._finalize_submission(raw_future, worker, self_ref, _holder)
+
+    def _finalize_submission(
+        self,
+        raw_future: Future,
+        worker,
+        self_ref,
+        holder: list,
+        ready: Optional[Event] = None,
+    ) -> TaskFuture:
+        task_future = TaskFuture(raw_future, worker)
+        # Store the TaskFuture in a mutable list so the _run() closure can see it after _ready is set.
+        holder[0] = task_future
+        if ready is not None:
+            ready.set()
+
+        # submitted fires before add_done_callback so it is always the first
+        # signal — even when the task finishes so fast that add_done_callback
+        # calls the callback synchronously in this thread.
+        self.submitted.emit(task_future)
+        raw_future.add_done_callback(
+            lambda f: self._done_callback(self_ref, f, task_future)
+        )
+        return task_future
+
+    def _get_manager(self):
+        if self._manager is None:
+            # SyncManager server process provides proxy objects that are
+            # picklable and safe to pass through ProcessPoolExecutor's pickle
+            # serialisation.  A plain multiprocessing.Queue is NOT picklable.
+            self._manager = multiprocessing.Manager()
+        return self._manager
+
+    def _handle_done(self, raw_future: Future, task_future: TaskFuture) -> None:
+        with self._lock:
+            self._running -= 1
+
+        if raw_future.cancelled():
+            return
+
+        if task_future.aborted():
+            self.aborted.emit(task_future)
+
+        try:
+            result = raw_future.result()
+        except Exception as exc:
+            self.failed.emit(task_future, exc)
+        else:
+            self.succeeded.emit(task_future, result)
+        finally:
+            self.finished.emit(task_future)
+
+    def shutdown(self, wait: bool = False) -> None:
+        self._executor.shutdown(wait=wait)
+        if self._manager is not None:
+            self._manager.shutdown()
+            self._manager = None
+
+    @staticmethod
+    def _done_callback(
+        executor_ref, raw_future: Future, task_future: TaskFuture
+    ) -> None:
+        executor = executor_ref()
+        if executor is None:
+            return
+        executor._handle_done(raw_future, task_future)
+
+    @staticmethod
+    def _emit_started(self_ref, holder: list) -> None:
+        exe = self_ref()
+        if exe is not None:
+            exe.started.emit(holder[0])
diff --git a/src/ewoksorange/gui/concurrency/executor/TaskFuture.py b/src/ewoksorange/gui/concurrency/executor/TaskFuture.py
new file mode 100644
index 00000000..258a8b59
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/TaskFuture.py
@@ -0,0 +1,41 @@
+from concurrent.futures import Future
+
+from ._EwoksTaskHandle import EwoksTaskHandle as _EwoksTaskHandle
+
+
+class TaskFuture:
+    """Wraps a concurrent.futures.Future with ewoks-specific abort support."""
+
+    def __init__(self, raw_future: Future, worker: _EwoksTaskHandle):
+        self._future = raw_future
+        self._worker = worker
+
+    def cancel(self) -> bool:
+        """Prevent execution if the task has not started (native future cancel)."""
+        return self._future.cancel()
+
+    def abort(self) -> bool:
+        """Abort a running ewoks task. Returns True if the task was reached and cancelled."""
+        return self._worker.abort()
+
+    def aborted(self) -> bool:
+        """Return True if the underlying ewoks task was aborted."""
+        return self._worker.aborted()
+
+    def cancelled(self) -> bool:
+        return self._future.cancelled()
+
+    def running(self) -> bool:
+        return self._future.running()
+
+    def done(self) -> bool:
+        return self._future.done()
+
+    def result(self, timeout=None):
+        return self._future.result(timeout=timeout)
+
+    def exception(self, timeout=None):
+        return self._future.exception(timeout=timeout)
+
+    def add_done_callback(self, fn):
+        self._future.add_done_callback(fn)
diff --git a/src/ewoksorange/gui/concurrency/executor/_EwoksProcessHandle.py b/src/ewoksorange/gui/concurrency/executor/_EwoksProcessHandle.py
new file mode 100644
index 00000000..489f49e5
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/_EwoksProcessHandle.py
@@ -0,0 +1,16 @@
+from ._EwoksTaskHandle import EwoksTaskHandle
+
+
+class EwoksProcessHandle(EwoksTaskHandle):
+    """Controls a task running in a subprocess via multiprocessing.Event objects."""
+
+    def __init__(self, abort_event, aborted_event):
+        self._abort_event = abort_event
+        self._aborted_event = aborted_event
+
+    def abort(self) -> bool:
+        self._abort_event.set()
+        return True
+
+    def aborted(self) -> bool:
+        return self._aborted_event.is_set()
diff --git a/src/ewoksorange/gui/concurrency/executor/_EwoksTaskHandle.py b/src/ewoksorange/gui/concurrency/executor/_EwoksTaskHandle.py
new file mode 100644
index 00000000..fc502f54
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/_EwoksTaskHandle.py
@@ -0,0 +1,10 @@
+class EwoksTaskHandle:
+    """Common interface for objects that control a running ewoks task."""
+
+    def abort(self) -> None:
+        """Abort the running ewoks task."""
+        raise NotImplementedError("Base class")
+
+    def aborted(self) -> bool:
+        """Return True if the task was aborted."""
+        raise NotImplementedError("Base class")
diff --git a/src/ewoksorange/gui/concurrency/executor/_EwoksThreadHandle.py b/src/ewoksorange/gui/concurrency/executor/_EwoksThreadHandle.py
new file mode 100644
index 00000000..e63284b6
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/_EwoksThreadHandle.py
@@ -0,0 +1,46 @@
+from threading import Lock
+from typing import Optional
+from typing import Type
+
+from ewokscore import TaskWithProgress
+from ewokscore.task import Task
+
+from ._EwoksTaskHandle import EwoksTaskHandle
+
+
+class EwoksThreadHandle(EwoksTaskHandle):
+    """Callable that instantiates and executes an ewoks task in the worker thread."""
+
+    def __init__(self, task_class: Type[Task], **task_kwargs):
+        self._task_class = task_class
+        self._task_kwargs = task_kwargs
+        self._task: Optional[Task] = None
+        self._lock = Lock()
+        # Guard access to self._task
+
+    def __call__(self):
+        task_class = self._task_class
+        kwargs = dict(self._task_kwargs)
+        if not issubclass(task_class, TaskWithProgress):
+            kwargs.pop("progress", None)
+
+        task = task_class(**kwargs)
+        with self._lock:
+            self._task = task
+
+        task.execute()
+        return task.output_variables
+
+    def abort(self) -> bool:
+        """Call the ewoks task's cancel() to stop a running task. Returns True if the task was reached."""
+        with self._lock:
+            task = self._task
+            if task is not None:
+                task.cancel()
+                return True
+            return False
+
+    def aborted(self) -> bool:
+        with self._lock:
+            task = self._task
+        return task is not None and task.cancelled
diff --git a/src/ewoksorange/gui/concurrency/executor/_ProcessCallable.py b/src/ewoksorange/gui/concurrency/executor/_ProcessCallable.py
new file mode 100644
index 00000000..914b89f4
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/_ProcessCallable.py
@@ -0,0 +1,71 @@
+import threading
+from typing import Type
+
+from ewokscore.task import Task
+
+
+class ProcessCallable:
+    """Top-level picklable callable submitted to ProcessPoolExecutor.
+
+    IPC is handled by three objects passed at construction time:
+    - started_queue: multiprocessing.Queue — worker puts "started" when running
+    - abort_event:   multiprocessing.Event — main process sets to request cancel
+    - aborted_event: multiprocessing.Event — worker sets once the task has
+                     actually stopped running as a result of that request
+    """
+
+    def __init__(
+        self,
+        task_class: Type[Task],
+        task_kwargs: dict,
+        started_queue,
+        abort_event,
+        aborted_event,
+    ):
+        self._task_class = task_class
+        self._task_kwargs = task_kwargs
+        self._started_queue = started_queue
+        self._abort_event = abort_event
+        self._aborted_event = aborted_event
+
+    def __call__(self):
+        task_class = self._task_class
+        kwargs = dict(self._task_kwargs)
+        task = task_class(**kwargs)
+
+        # A daemon thread watches for the abort signal and calls task.cancel().
+        # `done` prevents cancel() from being called after the task has already
+        # finished naturally. Once the task actually stops (execute() returns),
+        # aborted_event is set so the main process can wait for abort completion.
+        done = threading.Event()
+
+        def _watch_abort():
+            try:
+                self._abort_event.wait()
+                if done.is_set():
+                    # abort_event was only set to release this thread.
+                    return
+                task.cancel()
+                done.wait()
+                self._aborted_event.set()
+            except (EOFError, BrokenPipeError, ConnectionError):
+                # The manager providing the event proxies shut down.
+                pass
+
+        watcher = threading.Thread(target=_watch_abort, daemon=True)
+        watcher.start()
+
+        # Signal "started" after the task and watcher are set up so that an
+        # abort() arriving right after started cannot race with reset_state().
+        self._started_queue.put("started")
+
+        try:
+            task.execute()
+        finally:
+            done.set()
+            # '_watch_abort' is waiting over the '_abort_event'. Release it.
+            # If done is already set will just release the thread.
+            self._abort_event.set()
+            watcher.join(timeout=5)
+
+        return task.output_variables
diff --git a/src/ewoksorange/gui/concurrency/executor/__init__.py b/src/ewoksorange/gui/concurrency/executor/__init__.py
new file mode 100644
index 00000000..1fb2bb37
--- /dev/null
+++ b/src/ewoksorange/gui/concurrency/executor/__init__.py
@@ -0,0 +1,3 @@
+from .EwoksExecutor import EwoksExecutor  # noqa F401
+from .EwoksExecutor import SubmitPolicy  # noqa F401
+from .TaskFuture import TaskFuture  # noqa F401
diff --git a/src/ewoksorange/tests/test_executor.py b/src/ewoksorange/tests/test_executor.py
new file mode 100644
index 00000000..44decc82
--- /dev/null
+++ b/src/ewoksorange/tests/test_executor.py
@@ -0,0 +1,223 @@
+"""Tests for EwoksExecutor."""
+
+import time
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from ewokscore import Task
+from ewokscore.missing_data import MissingData
+from ewokscore.tests.examples.tasks.sumtask import SumTask as _SumTask
+
+from ewoksorange.gui.concurrency.executor.EwoksExecutor import EwoksExecutor
+from ewoksorange.gui.concurrency.executor.EwoksExecutor import SubmitPolicy
+from ewoksorange.gui.qt_utils.app import QtEvent
+
+
+class SumTask(_SumTask):
+    def cancel(self):
+        pass
+
+
+class FailTask(Task, input_names=[], output_names=[]):
+    def run(self):
+        raise RuntimeError("deliberate failure")
+
+    def cancel(self):
+        pass
+
+
+class SleepTask(Task, input_names=("duration",), output_names=("result",)):
+    """Long-running task; abort() will cut it short by setting _cancelled."""
+
+    def run(self):
+        step = 0.05
+        elapsed = 0.0
+        while elapsed < self.inputs.duration:
+            if self.cancelled:
+                return
+            time.sleep(step)
+            elapsed += step
+        self.outputs.result = f"slept {elapsed:.2f}s"
+
+    def cancel(self):
+        self._cancelled = True
+
+
+def _output_values(output_variables) -> dict:
+    return {k: v.value for k, v in output_variables.items()}
+
+
+def _make_executor(policy=SubmitPolicy.ALWAYS, workers=1):
+    return EwoksExecutor(ThreadPoolExecutor(max_workers=workers), policy)
+
+
+@pytest.fixture(
+    params=[
+        pytest.param(
+            (ThreadPoolExecutor, 1, SubmitPolicy.DROP_IF_BUSY), id="thread-1-drop"
+        ),
+        pytest.param((ThreadPoolExecutor, 1, SubmitPolicy.ALWAYS), id="thread-1-queue"),
+        pytest.param(
+            (ThreadPoolExecutor, 4, SubmitPolicy.ALWAYS), id="thread-4-parallel"
+        ),
+        pytest.param(
+            (ProcessPoolExecutor, 2, SubmitPolicy.ALWAYS), id="process-2-parallel"
+        ),
+    ]
+)
+def executor(request, qtapp):
+    PoolClass, workers, policy = request.param
+    exe = EwoksExecutor(PoolClass(max_workers=workers), policy)
+    yield exe
+    exe.shutdown(wait=False)
+
+
+def test_submit_task_succeeded(executor):
+    result_holder = {}
+    done = QtEvent()
+
+    executor.succeeded.connect(
+        lambda task_future, result: (
+            result_holder.update({"result": result}),
+            done.set(),
+        )
+    )
+
+    task_future = executor.submit_task(SumTask, inputs={"a": 3, "b": 4})
+    assert task_future is not None
+    assert done.wait(timeout=5)
+    assert _output_values(result_holder["result"]) == {"result": 7}
+
+
+def test_submit_task_failed(executor):
+    exc_holder = {}
+    done = QtEvent()
+
+    executor.failed.connect(
+        lambda task_future, exception: (
+            exc_holder.update({"exception": exception}),
+            done.set(),
+        )
+    )
+
+    task_future = executor.submit_task(FailTask)
+    assert task_future is not None
+    assert done.wait(timeout=5)
+    assert isinstance(exc_holder["exception"], RuntimeError)
+    assert "deliberate failure" in str(exc_holder["exception"])
+
+
+def test_abort_running_task(executor):
+    """abort() exits the task early; succeeded and aborted both fire."""
+    started = QtEvent()
+    done = QtEvent()
+    result_holder = {}
+    aborted_holder = {}
+
+    executor.started.connect(lambda task_future: started.set())
+    executor.succeeded.connect(
+        lambda task_future, result: (
+            result_holder.update({"result": result}),
+            done.set(),
+        )
+    )
+    executor.failed.connect(
+        lambda task_future, exception: (
+            result_holder.update({"exception": exception}),
+            done.set(),
+        )
+    )
+    executor.aborted.connect(
+        lambda task_future: aborted_holder.update({"task_future": task_future})
+    )
+
+    task_future = executor.submit_task(SleepTask, inputs={"duration": 60.0})
+    assert started.wait(timeout=5), "task never started"
+
+    task_future.abort()
+
+    assert done.wait(timeout=5)
+    assert "result" in result_holder, f"expected succeeded, got: {result_holder}"
+    assert isinstance(_output_values(result_holder["result"])["result"], MissingData)
+    assert aborted_holder.get("task_future") is task_future
+
+
+def test_submitted_returns_task_future(executor):
+    done = QtEvent()
+
+    executor.succeeded.connect(lambda task_future, result: done.set())
+
+    task_future = executor.submit_task(SumTask, inputs={"a": 1, "b": 1})
+    assert task_future is not None
+    assert done.wait(timeout=5)
+
+
+def test_drop_if_busy_policy(qtapp):
+    """DROP_IF_BUSY silently ignores submissions while the executor is busy."""
+    submitted_count = [0]
+    ignored_count = [0]
+    done = QtEvent()
+
+    exe = _make_executor(policy=SubmitPolicy.DROP_IF_BUSY, workers=1)
+    exe.submitted.connect(
+        lambda task_future: submitted_count.__setitem__(0, submitted_count[0] + 1)
+    )
+    exe.ignored.connect(lambda: ignored_count.__setitem__(0, ignored_count[0] + 1))
+    exe.succeeded.connect(lambda task_future, result: done.set())
+
+    # First submit starts, the rest should be dropped
+    for _ in range(5):
+        exe.submit_task(SleepTask, inputs={"duration": 0.3})
+
+    assert done.wait(timeout=10)
+    assert submitted_count[0] == 1
+    assert ignored_count[0] == 4
+    exe.shutdown()
+
+
+def test_cancel_queued_task(qtapp):
+    """cancel() on a pending future returns True and prevents execution.
+
+    Requires ALWAYS policy with a single worker so the second task queues
+    behind the first rather than starting immediately or being dropped.
+    """
+    done = QtEvent()
+    second_ran = [False]
+
+    exe = _make_executor(policy=SubmitPolicy.ALWAYS, workers=1)
+    exe.succeeded.connect(lambda task_future, result: done.set())
+
+    # Block the single worker thread
+    exe.submit_task(SleepTask, inputs={"duration": 2.0})
+    time.sleep(0.1)  # ensure the blocker has started
+
+    # This one is pending — cancel it before it runs
+    tf = exe.submit_task(SumTask, inputs={"a": 1, "b": 2})
+    exe.succeeded.connect(lambda task_future, result: second_ran.__setitem__(0, True))
+    cancelled = tf.cancel()
+
+    assert cancelled is True
+    assert not second_ran[0]
+    exe.shutdown(wait=False)
+
+
+def test_multiple_parallel_tasks(qtapp):
+    """Four tasks complete concurrently when given four worker threads."""
+    results = []
+    done = QtEvent()
+
+    exe = _make_executor(policy=SubmitPolicy.ALWAYS, workers=4)
+    exe.succeeded.connect(
+        lambda task_future, result: (
+            results.append(_output_values(result)["result"]),
+            (done.set() if len(results) == 4 else None),
+        )
+    )
+
+    for i in range(4):
+        exe.submit_task(SumTask, inputs={"a": i, "b": i})
+
+    assert done.wait(timeout=10)
+    assert sorted(results) == [0, 2, 4, 6]
+    exe.shutdown()