From 6837feb362bb25127d0272291f53d7247a5c93d6 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Tue, 11 Aug 2026 16:39:23 +0100 Subject: [PATCH 1/6] POC Resuming tasks --- src/blueapi/worker/task.py | 1 + src/blueapi/worker/task_worker.py | 36 ++++++++++++++++++++++++------- 2 files changed, 29 insertions(+), 8 deletions(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 9ce373c769..50bfe93c6a 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -39,6 +39,7 @@ def do_task(self, ctx: BlueskyContext) -> None: prepared_params = self.prepare_params(ctx) ctx.run_engine.md.update(self.metadata) result = ctx.run_engine(func(**prepared_params)) + if isinstance(result, tuple): # pragma: no cover # this is never true if the run_engine is configured correctly return None diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index 4ada3c2a54..c1d0f97b51 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -10,6 +10,7 @@ from bluesky._vendor.super_state_machine.errors import TransitionError from bluesky.protocols import Status +from bluesky.utils import RunEngineInterrupted from observability_utils.tracing import ( add_span_attributes, get_tracer, @@ -75,9 +76,11 @@ class TrackableTask(BlueapiBaseModel): def set_result(self, result: Any): self.outcome = TaskResult.from_result(result) + self.is_complete = True def set_exception(self, err: Exception): self.outcome = TaskError.from_exception(err) + self.is_complete = True class TaskWorker: @@ -315,8 +318,9 @@ def mark_task_as_started(event: WorkerEvent, _: str | None) -> None: self._current_task_otel_context = get_current() """ Cache the current trace context as the one for this task id """ self._task_channel.put_nowait(trackable_task) - task_started.wait(timeout=5.0) - if not task_started.is_set(): + if task_started.wait(timeout=5.0): + self._pending_tasks.pop(trackable_task.task_id) + else: raise TimeoutError("Failed to start plan within timeout") except Full as f: LOGGER.error("Cannot submit task while another is running") @@ -410,7 +414,8 @@ def resume(self): Command the worker to resume """ LOGGER.info("Requesting to resume the worker") - self._ctx.run_engine.resume() + self._current_task_otel_context = get_current() + self._task_channel.put(ResumeSignal()) @start_as_current_span(TRACER) def _cycle_with_error_handling(self) -> None: @@ -423,20 +428,34 @@ def _cycle_with_error_handling(self) -> None: def _cycle(self) -> None: try: LOGGER.info("Awaiting task") - next_task: TrackableTask | KillSignal = self._task_channel.get() + next_task: TrackableTask | KillSignal | ResumeSignal = ( + self._task_channel.get() + ) + if isinstance(next_task, ResumeSignal): + if self._current is None or self._current.is_complete: + raise Exception("Can't resume") + next_task = self._current if isinstance(next_task, TrackableTask): def process_task(): LOGGER.info(f"Got new task: {next_task}") self._current = next_task + resume = not self._current.is_pending self._current.is_pending = False meta = {"task_id": self._current.task_id} try: - result = self._current.task.do_task(self._ctx) + if resume: + result = self._ctx.run_engine.resume() + else: + result = self._current.task.do_task(self._ctx) LOGGER.info( "Task ran successfully - returned: %s", result, extra=meta ) self._current.set_result(result) + except RunEngineInterrupted: + LOGGER.info("Task paused") + if self._ctx.run_engine.state != "paused": + raise except Exception as e: LOGGER.error("Task failed", extra=meta) self._current.set_exception(e) @@ -472,9 +491,7 @@ def process_task(): if self._current_task_otel_context is not None: self._current_task_otel_context = None - if self._current is not None: - self._current.is_complete = True - self._pending_tasks.pop(self._current.task_id) + if self._current is not None and self._current.is_complete: self._completed_tasks[self._current.task_id] = self._current self._report_status() self._errors.clear() @@ -683,6 +700,9 @@ class KillSignal: ... +class ResumeSignal: ... + + def run_worker_in_own_thread( worker: TaskWorker, executor: ThreadPoolExecutor | None = None ) -> Future: From 20edaeed90d01bf2844f3ac69cdad9ec91219aec Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Tue, 11 Aug 2026 16:45:24 +0100 Subject: [PATCH 2/6] Fix task results --- src/blueapi/worker/task_worker.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index c1d0f97b51..6d86c3ad41 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -10,6 +10,7 @@ from bluesky._vendor.super_state_machine.errors import TransitionError from bluesky.protocols import Status +from bluesky.run_engine import RunEngineResult from bluesky.utils import RunEngineInterrupted from observability_utils.tracing import ( add_span_attributes, @@ -446,12 +447,14 @@ def process_task(): try: if resume: result = self._ctx.run_engine.resume() + if isinstance(result, RunEngineResult): + self._current.set_result(result.plan_result) else: result = self._current.task.do_task(self._ctx) + self._current.set_result(result) LOGGER.info( "Task ran successfully - returned: %s", result, extra=meta ) - self._current.set_result(result) except RunEngineInterrupted: LOGGER.info("Task paused") if self._ctx.run_engine.state != "paused": From bbf005e7f32f5c6d760445056b37324fb9d69822 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Tue, 11 Aug 2026 17:15:06 +0100 Subject: [PATCH 3/6] Add logging and error handling --- src/blueapi/worker/task.py | 10 +++------- src/blueapi/worker/task_worker.py | 27 +++++++++++++++++---------- 2 files changed, 20 insertions(+), 17 deletions(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 50bfe93c6a..5497024054 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -2,6 +2,7 @@ from collections.abc import Mapping from typing import Any +from bluesky.run_engine import RunEngineResult from pydantic import BaseModel, Field, TypeAdapter from blueapi.core import BlueskyContext @@ -29,7 +30,7 @@ def prepare_params(self, ctx: BlueskyContext) -> Mapping[str, Any]: # Re-create dict manually to avoid nesting in model_dump output return {field: getattr(model, field) for field in model.__pydantic_fields__} - def do_task(self, ctx: BlueskyContext) -> None: + def do_task(self, ctx: BlueskyContext) -> RunEngineResult | tuple[str, ...]: LOGGER.info( f"Asked to run plan {self.name} with {self.params} and " f"metadata {self.metadata} for all runs" @@ -38,12 +39,7 @@ def do_task(self, ctx: BlueskyContext) -> None: func = ctx.plan_functions[self.name] prepared_params = self.prepare_params(ctx) ctx.run_engine.md.update(self.metadata) - result = ctx.run_engine(func(**prepared_params)) - - if isinstance(result, tuple): # pragma: no cover - # this is never true if the run_engine is configured correctly - return None - return result.plan_result + return ctx.run_engine(func(**prepared_params)) def _lookup_params(ctx: BlueskyContext, task: Task) -> BaseModel: diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index 6d86c3ad41..baa3b1c55f 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -434,29 +434,36 @@ def _cycle(self) -> None: ) if isinstance(next_task, ResumeSignal): if self._current is None or self._current.is_complete: - raise Exception("Can't resume") + LOGGER.debug( + "Ignoring resume signal - nothing to resume: current=%s", + self._current, + ) + return + # If there is an incomplete task, treat it as the next task to run next_task = self._current + if isinstance(next_task, TrackableTask): def process_task(): LOGGER.info(f"Got new task: {next_task}") self._current = next_task - resume = not self._current.is_pending - self._current.is_pending = False meta = {"task_id": self._current.task_id} try: - if resume: - result = self._ctx.run_engine.resume() - if isinstance(result, RunEngineResult): - self._current.set_result(result.plan_result) - else: + if self._current.is_pending: + self._current.is_pending = False result = self._current.task.do_task(self._ctx) - self._current.set_result(result) + else: + LOGGER.debug("Resuming previous task") + result = self._ctx.run_engine.resume() + if isinstance(result, RunEngineResult): + # Should always be a RunEngineResult if the run + # engine is configured correctly + self._current.set_result(result.plan_result) LOGGER.info( "Task ran successfully - returned: %s", result, extra=meta ) except RunEngineInterrupted: - LOGGER.info("Task paused") + LOGGER.info("Task paused", extra=meta) if self._ctx.run_engine.state != "paused": raise except Exception as e: From cbda075a615841f9d238d299858ddc20f04bc4f9 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Wed, 12 Aug 2026 11:06:00 +0100 Subject: [PATCH 4/6] Restore log message --- src/blueapi/worker/task_worker.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index baa3b1c55f..e731467291 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -459,9 +459,16 @@ def process_task(): # Should always be a RunEngineResult if the run # engine is configured correctly self._current.set_result(result.plan_result) - LOGGER.info( - "Task ran successfully - returned: %s", result, extra=meta - ) + LOGGER.info( + "Task ran successfully - returned: %s", + result.plan_result, + extra=meta, + ) + else: + LOGGER.warn( + "Task ran successfully but did not return " + "RunEngineResult. Is `call_returns_result` set?" + ) except RunEngineInterrupted: LOGGER.info("Task paused", extra=meta) if self._ctx.run_engine.state != "paused": From 8fd1f3ab131bfc78cc59f5d10c60e15e82b92c28 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Wed, 12 Aug 2026 15:19:51 +0100 Subject: [PATCH 5/6] Add state locks and abort signals --- docs/reference/openapi.yaml | 5 +- src/blueapi/config.py | 2 +- src/blueapi/worker/task_worker.py | 107 ++++++++++++++------ tests/unit_tests/service/test_rest_api.py | 12 +-- tests/unit_tests/worker/test_task_worker.py | 24 ++--- 5 files changed, 94 insertions(+), 56 deletions(-) diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index e0cd8282df..9182125167 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -365,7 +365,7 @@ components: title: Errors type: array is_complete: - default: false + readOnly: true title: Is Complete type: boolean is_pending: @@ -389,6 +389,7 @@ components: required: - task_id - task + - is_complete title: TrackableTask type: object ValidationError: @@ -449,7 +450,7 @@ info: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html title: BlueAPI Control - version: 1.5.0 + version: 1.5.1 openapi: 3.1.0 paths: /api/v1/devices: diff --git a/src/blueapi/config.py b/src/blueapi/config.py index a181f4c344..b93704c57b 100644 --- a/src/blueapi/config.py +++ b/src/blueapi/config.py @@ -324,7 +324,7 @@ class ApplicationConfig(BlueapiBaseModel): """ #: API version to publish in OpenAPI schema - REST_API_VERSION: ClassVar[str] = "1.5.0" + REST_API_VERSION: ClassVar[str] = "1.5.1" LICENSE_INFO: ClassVar[dict[str, str]] = { "name": "Apache 2.0", diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index e731467291..fa0c6f3360 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from functools import partial from queue import Full, Queue -from threading import Event, RLock +from threading import Condition, Event, RLock from typing import Any, TypeVar from bluesky._vendor.super_state_machine.errors import TransitionError @@ -21,7 +21,7 @@ from opentelemetry.baggage import get_baggage from opentelemetry.context import Context, get_current from opentelemetry.trace import SpanKind -from pydantic import Field +from pydantic import Field, computed_field, model_validator from pydantic.json_schema import SkipJsonSchema from blueapi.core import ( @@ -70,18 +70,27 @@ class TrackableTask(BlueapiBaseModel): task_id: str task: Task request_id: str | SkipJsonSchema[None] = None - is_complete: bool = False is_pending: bool = True errors: list[str] = Field(default_factory=list) outcome: TaskResult | TaskError | None = None + @computed_field + @property + def is_complete(self) -> bool: + return self.outcome is not None + + @model_validator(mode="before") + @classmethod + def remove_complete(cls, values: dict[str, Any]) -> dict[str, Any]: + # fix pydantic falling over itself with computed fields + values.pop("is_complete", None) + return values + def set_result(self, result: Any): self.outcome = TaskResult.from_result(result) - self.is_complete = True def set_exception(self, err: Exception): self.outcome = TaskError.from_exception(err) - self.is_complete = True class TaskWorker: @@ -114,6 +123,7 @@ class TaskWorker: _task_channel: Queue # type: ignore _current: TrackableTask | None _status_lock: RLock + _state_change: Condition _status_snapshot: dict[str, StatusView] _completed_statuses: set[str] _worker_events: EventPublisher[WorkerEvent] @@ -147,6 +157,7 @@ def __init__( self._progress_events = EventPublisher() self._data_events = EventPublisher() self._status_lock = RLock() + self._state_change = Condition() self._status_snapshot = {} self._completed_statuses = set() self._started = Event() @@ -184,22 +195,32 @@ def cancel_active_task( Returns: The task_id of the active task """ - if self._current is None: + + current = self._current + if current is None: # Persuades type checker that self._current is not None # We only allow this method to be called if a Plan is active raise TransitionError("Attempted to cancel while no active Task") - if failure: - default_reason = "Task failed for unknown reason" - self._ctx.run_engine.abort(reason or default_reason) - add_span_attributes({"Task aborted": reason or default_reason}) + + if self._current_task_otel_context is None: + self._current_task_otel_context = get_current() + + reason = reason or ("Task aborted" if failure else "Task stopped") + if self._state != WorkerState.RUNNING: + with self._state_change: + self._task_channel.put(AbortSignal(failure, reason)) + self._state_change.wait_for(lambda: self._state == WorkerState.IDLE) else: - self._ctx.run_engine.stop() - default_reason = "Cancellation successful: Task stopped without error" - add_span_attributes({"Task stopped": reason or default_reason}) - return self._current.task_id + if failure: + current.outcome = TaskError(type="Abort", message=reason) + self._ctx.run_engine.abort(reason=reason) + else: + current.outcome = TaskResult.from_result(None) + self._ctx.run_engine.stop() + return current.task_id @start_as_current_span(TRACER, "task_id") - def get_task_by_id(self, task_id: str) -> TrackableTask | None: + def get_task_by_id(self, task_id: str) -> TrackableTask: """ Returns a task matching the task ID supplied, if the worker knows of it. @@ -209,6 +230,9 @@ def get_task_by_id(self, task_id: str) -> TrackableTask | None: Optional[TrackableTask[T]]: The task matching the ID, None if the task ID is unknown to the worker. """ + current = self._current + if current and current.task_id == task_id: + return current return self._pending_tasks.get(task_id, None) or self._completed_tasks[task_id] @start_as_current_span(TRACER) @@ -222,11 +246,8 @@ def get_tasks(self, status: TaskStatusEnum | None = None) -> list[TrackableTask] list[TrackableTask]: A list of tasks that match the given status. """ if status == TaskStatusEnum.RUNNING: - return [ - task - for task in self._pending_tasks.values() - if not task.is_pending and not task.is_complete - ] + current = self._current + return [current] if current else [] elif status == TaskStatusEnum.PENDING: return [task for task in self._pending_tasks.values() if task.is_pending] elif status == TaskStatusEnum.COMPLETE: @@ -416,7 +437,9 @@ def resume(self): """ LOGGER.info("Requesting to resume the worker") self._current_task_otel_context = get_current() - self._task_channel.put(ResumeSignal()) + with self._state_change: + self._task_channel.put(ResumeSignal()) + self._state_change.wait_for(lambda: self._state != WorkerState.PAUSED) @start_as_current_span(TRACER) def _cycle_with_error_handling(self) -> None: @@ -429,7 +452,7 @@ def _cycle_with_error_handling(self) -> None: def _cycle(self) -> None: try: LOGGER.info("Awaiting task") - next_task: TrackableTask | KillSignal | ResumeSignal = ( + next_task: TrackableTask | KillSignal | ResumeSignal | AbortSignal = ( self._task_channel.get() ) if isinstance(next_task, ResumeSignal): @@ -470,9 +493,7 @@ def process_task(): "RunEngineResult. Is `call_returns_result` set?" ) except RunEngineInterrupted: - LOGGER.info("Task paused", extra=meta) - if self._ctx.run_engine.state != "paused": - raise + LOGGER.info("Task interrupted", extra=meta) except Exception as e: LOGGER.error("Task failed", extra=meta) self._current.set_exception(e) @@ -493,6 +514,16 @@ def process_task(): process_task() else: process_task() + elif isinstance(next_task, AbortSignal): + if self._current and not self._current.is_complete: + if next_task.failure: + result = self._ctx.run_engine.abort(reason=next_task.reason) + self._current.outcome = TaskError( + type="Abort", message=next_task.reason + ) + else: + result = self._ctx.run_engine.stop() + self._current.set_result(result.plan_result) elif isinstance(next_task, KillSignal): # If we receive a kill signal we begin to shut the worker down. @@ -548,14 +579,18 @@ def _on_state_change( raw_new_state: RawRunEngineState, raw_old_state: RawRunEngineState | None = None, ) -> None: - new_state = WorkerState.from_bluesky_state(raw_new_state) - if raw_old_state: - old_state = WorkerState.from_bluesky_state(raw_old_state) - else: - old_state = WorkerState.UNKNOWN - LOGGER.debug(f"Notifying state change {old_state} -> {new_state}") - self._state = new_state - self._report_status() + print("Start of _on_state_change") + with self._state_change: + new_state = WorkerState.from_bluesky_state(raw_new_state) + if raw_old_state: + old_state = WorkerState.from_bluesky_state(raw_old_state) + else: + old_state = WorkerState.UNKNOWN + LOGGER.debug(f"Notifying state change {old_state} -> {new_state}") + self._state = new_state + self._state_change.notify() + self._report_status() + print("End of _on_state_change") def _report_error(self, err: Exception) -> None: LOGGER.error(err, exc_info=True) @@ -720,6 +755,12 @@ class KillSignal: class ResumeSignal: ... +@dataclass +class AbortSignal: + failure: bool + reason: str + + def run_worker_in_own_thread( worker: TaskWorker, executor: ThreadPoolExecutor | None = None ) -> Future: diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index dedd9d7ff3..16bc4e2970 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -45,7 +45,7 @@ from blueapi.service.runner import WorkerDispatcher from blueapi.worker.event import WorkerState from blueapi.worker.task import Task -from blueapi.worker.task_worker import TrackableTask +from blueapi.worker.task_worker import TaskResult, TrackableTask class MockCountModel(BaseModel): ... @@ -371,7 +371,7 @@ def test_put_plan_fails_if_not_idle(mock_runner: Mock, client: TestClient) -> No # Set to non idle mock_runner.run.return_value = TrackableTask( - task=Task(name="none"), task_id=task_id_current, is_complete=False + task=Task(name="none"), task_id=task_id_current ) resp = client.put("/worker/task", json={"task_id": task_id_new}) @@ -386,7 +386,6 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: TrackableTask( task_id="1", task=Task(name="first_task"), - is_complete=False, is_pending=True, ), ] @@ -433,7 +432,7 @@ def test_get_tasks_by_status(mock_runner: Mock, client: TestClient) -> None: TrackableTask( task_id="3", task=Task(name="third_task"), - is_complete=True, + outcome=TaskResult.from_result(42), is_pending=False, ), ] @@ -453,7 +452,7 @@ def test_get_tasks_by_status(mock_runner: Mock, client: TestClient) -> None: "params": {}, "metadata": {}, }, - "outcome": None, + "outcome": {"outcome": "success", "type": "int", "result": 42}, "task_id": "3", } ] @@ -534,7 +533,7 @@ def test_set_active_task_active_task_complete( mock_runner.run.return_value = TrackableTask( task_id="1", task=Task(name="a_completed_task"), - is_complete=True, + outcome=TaskResult.from_result(42), is_pending=False, ) @@ -553,7 +552,6 @@ def test_set_active_task_worker_already_running( mock_runner.run.return_value = TrackableTask( task_id="1", task=Task(name="a_running_task"), - is_complete=False, is_pending=False, ) diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 588c37d8e4..a3811ac5a6 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -53,6 +53,8 @@ }, ) +_SUCCESS = TaskResult(type="int", result=42) + class FakeDevice(Movable[float]): event: threading.Event @@ -629,21 +631,19 @@ def callback(unused: Future[list[Any]], stream=stream, sub=sub): ], ) def test_get_tasks(worker: TaskWorker, status, expected_task_ids): + worker._current = TrackableTask( + task_id="task1", + task=Task(name="set_absolute", params={"movable": "fake_device", "value": 4.0}), + outcome=None, + is_pending=False, + ) worker._pending_tasks = { - "task1": TrackableTask( - task_id="task1", - task=Task( - name="set_absolute", params={"movable": "fake_device", "value": 4.0} - ), - is_complete=False, - is_pending=False, - ), "task2": TrackableTask( task_id="task2", task=Task( name="set_absolute", params={"movable": "fake_device", "value": 4.0} ), - is_complete=False, + outcome=_SUCCESS, is_pending=True, ), } @@ -653,7 +653,7 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): task=Task( name="set_absolute", params={"movable": "fake_device", "value": 4.0} ), - is_complete=True, + outcome=_SUCCESS, is_pending=False, ), } @@ -668,7 +668,7 @@ def test_submitting_completed_task_fails(worker: TaskWorker): with pytest.raises(ValueError): worker._submit_trackable_task( TrackableTask( - task_id="task1", task=_SIMPLE_TASK, is_complete=True, is_pending=False + task_id="task1", task=_SIMPLE_TASK, is_pending=False, outcome=_SUCCESS ) ) @@ -805,8 +805,6 @@ def test_cycle_without_otel_context(mock_logger: Mock, inert_worker: TaskWorker) assert inert_worker._current_task_otel_context is None # Bad way to tell that this branch has been run, but I can't think of a better way # Have to set these values to match output - task.is_complete = False - task.is_pending = True mock_logger.info.assert_called_with( "Task ran successfully - returned: %s", None, extra={"task_id": "0"} ) From 18f0b28674d3d4bcf32b472190348fdb3b143e54 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Thu, 13 Aug 2026 10:55:21 +0100 Subject: [PATCH 6/6] Report interruptions as errors --- src/blueapi/worker/task_worker.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index fa0c6f3360..046e18c142 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -492,8 +492,10 @@ def process_task(): "Task ran successfully but did not return " "RunEngineResult. Is `call_returns_result` set?" ) - except RunEngineInterrupted: + except RunEngineInterrupted as rei: LOGGER.info("Task interrupted", extra=meta) + if self._ctx.run_engine.state != "paused": + self._report_error(rei) except Exception as e: LOGGER.error("Task failed", extra=meta) self._current.set_exception(e)