fix: Convert ValidationErrors into something picklable

src/blueapi/service/main.py

@@ -57,6 +57,7 @@
 )
 from blueapi.worker import TrackableTask, WorkerState
 from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent
+from blueapi.worker.task import TaskParameterError
 from blueapi.worker.worker_errors import WorkerBusyError
 
 from .authorization import (
@@ -347,7 +348,7 @@ def submit_task(
         task_id: str = runner.run(interface.submit_task, task_request, {"user": fedid})
         response.headers["Location"] = f"{request.url}/{task_id}"
         return TaskResponse(task_id=task_id)
-    except ValidationError as e:
+    except TaskParameterError as e:
         # Add body/params context to location and ensure that all required
         # fields defined in the generated schema are present
         errors = [
@@ -358,7 +359,7 @@ def submit_task(
                 # Input is not listed as required but is useful to have if available
                 "input": err.get("input", None),
             }
-            for err in e.errors()
+            for err in e.errors
         ]
 
         LOGGER.info("Error submitting task: %s - %s", task_request, e)
@@ -669,9 +670,9 @@ async def run_plan(
             interface.submit_task, task_request.task, {"user": user}
         )
         LOGGER.info("Task ID: %s", task_id)
-    except ValidationError as ve:
-        LOGGER.info("Plan args not valid: %s - %s", task_request, ve)
-        await ws.send_text(InvalidArgs.from_validation_error(ve).model_dump_json())
+    except TaskParameterError as tpe:
+        LOGGER.info("Plan args not valid: %s - %s", task_request, tpe)
+        await ws.send_text(InvalidArgs.from_parameter_error(tpe).model_dump_json())
         await ws.close(code=protocol.INVALID_ARGS, reason="Invalid Args")
         return
     except KeyError as ke:

src/blueapi/service/protocol.py

@@ -17,11 +17,12 @@
 
 from typing import Annotated, Any, Literal, Self
 
-from pydantic import BaseModel, Field, TypeAdapter, ValidationError
+from pydantic import BaseModel, Field, TypeAdapter
 
 from blueapi.core.bluesky_types import DataEvent
 from blueapi.service.model import TaskRequest
 from blueapi.worker.event import ProgressEvent, WorkerEvent
+from blueapi.worker.task import TaskParameterError
 
 UNKNOWN_PLAN = 4001
 INVALID_ARGS = 4002
@@ -68,7 +69,7 @@ class InvalidArgs(BaseModel):
     errors: list[ArgumentError]
 
     @classmethod
-    def from_validation_error(cls, e: ValidationError) -> Self:
+    def from_parameter_error(cls, e: TaskParameterError) -> Self:
         errors = [
             ArgumentError(
                 loc=["body", "params", *err.get("loc", [])],
@@ -77,7 +78,7 @@ def from_validation_error(cls, e: ValidationError) -> Self:
                 # Input is not listed as required but is useful to have if available
                 input=err.get("input", None),
             )
-            for err in e.errors()
+            for err in e.errors
         ]
         return cls(errors=errors)
 

src/blueapi/worker/task.py

@@ -1,8 +1,8 @@
 import logging
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, Self, TypedDict
 
-from pydantic import BaseModel, Field, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter, ValidationError
 
 from blueapi.core import BlueskyContext
 from blueapi.utils import BlueapiBaseModel
@@ -25,9 +25,12 @@ class Task(BlueapiBaseModel):
     )
 
     def prepare_params(self, ctx: BlueskyContext) -> Mapping[str, Any]:
-        model = _lookup_params(ctx, self)
-        # Re-create dict manually to avoid nesting in model_dump output
-        return {field: getattr(model, field) for field in model.__pydantic_fields__}
+        try:
+            model = _lookup_params(ctx, self)
+            # Re-create dict manually to avoid nesting in model_dump output
+            return {field: getattr(model, field) for field in model.__pydantic_fields__}
+        except ValidationError as ve:
+            raise TaskParameterError.from_validation_error(ve) from None
 
     def do_task(self, ctx: BlueskyContext) -> None:
         LOGGER.info(
@@ -62,3 +65,34 @@ def _lookup_params(ctx: BlueskyContext, task: Task) -> BaseModel:
     model = plan.model
     adapter = TypeAdapter(model)
     return adapter.validate_python(task.params)
+
+
+class InvalidTaskParameter(TypedDict):
+    loc: tuple[str | int, ...]
+    msg: str | None
+    type: str | None
+    input: Any | None
+
+
+class TaskParameterError(ValueError):
+    errors: list[InvalidTaskParameter]
+
+    def __init__(self, message: str, errors: list[InvalidTaskParameter]):
+        super().__init__(message)
+        self.errors = errors
+
+    @classmethod
+    def from_validation_error(cls, error: ValidationError) -> Self:
+        errors = [
+            InvalidTaskParameter(
+                loc=e.get("loc", []),
+                msg=e.get("msg", None),
+                type=e.get("type", None),
+                input=e.get("input", None),
+            )
+            for e in error.errors()
+        ]
+        return cls(str(error), errors)
+
+    def __reduce__(self):
+        return (self.__class__, (*self.args, self.errors))

tests/system_tests/test_blueapi_system.py

@@ -18,7 +18,9 @@
     BlueapiRestClient,
     BlueskyRemoteControlError,
     BlueskyRequestError,
+    InvalidParametersError,
     NotFoundError,
+    ParameterError,
     ServiceUnavailableError,
     UnauthorisedAccessError,
     UnknownPlanError,
@@ -661,6 +663,73 @@ def test_stub_runs(client_with_stomp: BlueapiClient, task: TaskRequest):
     assert not final_event.task_failed
 
 
+@pytest.mark.parametrize(
+    "task,param_error,user",
+    [
+        (
+            TaskRequest(
+                name="count",
+                params={"detectors": "det", "num": 3},  # detectors should be list
+                instrument_session=VALID_INSTRUMENT_SESSION[User.alice],
+            ),
+            ParameterError(
+                loc=["body", "params", "detectors"],
+                msg="'str' instances are not allowed as a Sequence value",
+                type="sequence_str",
+                input="det",
+            ),
+            User.alice,
+        ),
+        (
+            TaskRequest(
+                name="spec_scan",
+                params={
+                    "detectors": ["det"],
+                    "spec": {
+                        "axis": "missing",  # doesn't exist
+                        "start": 0.0,
+                        "stop": 0.4,
+                        "num": 2,
+                        "type": "Linspace",
+                    },
+                },
+                instrument_session=VALID_INSTRUMENT_SESSION[User.bob],
+            ),
+            ParameterError(
+                loc=["body", "params", "spec", "Linspace", "axis"],
+                msg="Value error, Device missing cannot be found",
+                type="value_error",
+                input="missing",
+            ),
+            User.bob,
+        ),
+    ],
+)
+@pytest.mark.parametrize("run_method", ["run_task", "run_blocking"])
+def test_invalid_parameters_fail_cleanly(
+    client_with_stomp: BlueapiClient,
+    task: TaskRequest,
+    user: ValidUser,
+    param_error: ParameterError,
+    run_method: str,
+):
+    resource = Queue(maxsize=1)
+    start = Queue(maxsize=1)
+
+    def on_event(event: AnyEvent) -> None:
+        if isinstance(event, DataEvent):
+            if event.name == "start":
+                start.put_nowait(event.doc)
+            if event.name == "stream_resource":
+                resource.put_nowait(event.doc)
+
+    runner = getattr(client_with_stomp, run_method)
+    with pytest.raises(InvalidParametersError) as ipe:
+        runner(task, on_event)
+
+    assert ipe.value.errors == [param_error]
+
+
 # Regression test for #1480
 def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient):
     with pytest.raises(NotFoundError):

tests/unit_tests/service/test_protocol.py

@@ -1,8 +1,6 @@
 from typing import Any
 
 import pytest
-from pydantic import ValidationError
-from pydantic_core import InitErrorDetails
 
 from blueapi.service.model import TaskRequest
 from blueapi.service.protocol import (
@@ -15,6 +13,7 @@
     Resume,
     Submit,
 )
+from blueapi.worker.task import InvalidTaskParameter, TaskParameterError
 
 
 @pytest.mark.parametrize(
@@ -72,19 +71,20 @@ def test_response_deserialization(src: str, res: Any):
     assert req == res
 
 
-def test_from_empty_validation_error():
-    err = InvalidArgs.from_validation_error(
-        ValidationError("Error validating request", [])
+def test_from_empty_parameter_error():
+    err = InvalidArgs.from_parameter_error(
+        TaskParameterError("Error validating request", [])
     )
     assert err == InvalidArgs(errors=[])
 
 
-def test_from_validation_error():
-    err = InvalidArgs.from_validation_error(
-        ValidationError.from_exception_data(
-            title="Error validating request",
-            line_errors=[
-                InitErrorDetails(
+def test_from_parameter_error():
+    err = InvalidArgs.from_parameter_error(
+        TaskParameterError(
+            "Error validating request",
+            [
+                InvalidTaskParameter(
+                    msg="foo.bar is missing",
                     loc=("foo", "bar"),
                     type="missing",
                     input={"foo": {"no": "bar"}},
@@ -96,7 +96,7 @@ def test_from_validation_error():
         errors=[
             ArgumentError(
                 loc=["body", "params", "foo", "bar"],
-                msg="Field required",
+                msg="foo.bar is missing",
                 type="missing",
                 input={"foo": {"no": "bar"}},
             )

tests/unit_tests/service/test_rest_api.py

@@ -12,8 +12,7 @@
 from fastapi import FastAPI, HTTPException, WebSocketDisconnect, status
 from fastapi.testclient import TestClient
 from httpx2 import Headers
-from pydantic import BaseModel, ValidationError
-from pydantic_core import InitErrorDetails
+from pydantic import BaseModel
 from starlette.testclient import WebSocketDenialResponse
 from starlette.types import Message, Receive, Scope, Send
 
@@ -47,7 +46,7 @@
 )
 from blueapi.service.runner import WorkerDispatcher
 from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState
-from blueapi.worker.task import Task
+from blueapi.worker.task import InvalidTaskParameter, Task, TaskParameterError
 from blueapi.worker.task_worker import TrackableTask
 
 
@@ -330,21 +329,24 @@ def test_create_task_inserts_auth_metadata(
     task_id = str(uuid.uuid4())
 
     # mock_runner.run.side_effect = [task_id]
-    mock_runner.run.return_value = [task_id]
+    mock_runner.run.return_value = task_id
 
     client_with_auth.post("/tasks", json=task.model_dump())
 
     mock_runner.run.assert_called_with(submit_task, task, {"user": "jd1"})
 
 
-def test_create_task_validation_error(mock_runner: Mock, client: TestClient) -> None:
+def test_create_task_parameter_error(mock_runner: Mock, client: TestClient) -> None:
     mock_runner.run.side_effect = [
-        ValidationError.from_exception_data(
-            title="ValueError",
-            line_errors=[
-                InitErrorDetails(
-                    type="missing", loc=("id",), msg="value is required for Identifier"
-                )  # type: ignore
+        TaskParameterError(
+            "ValueError",
+            [
+                InvalidTaskParameter(
+                    type="missing",
+                    loc=("id",),
+                    msg="Field required",
+                    input=None,
+                )
             ],
         ),
     ]
@@ -1099,7 +1101,7 @@ def test_websocket_run_plan_invalid_request(
             "Unknown Plan",
         ),
         (
-            ValidationError("Not valid", []),
+            TaskParameterError("Not valid", []),
             {"kind": "invalid_args", "errors": []},
             4002,
             "Invalid Args",