[cross-repo from server#301] Conformance: prove malformed signal/query payload typed errors (#106)

Author: durable-workflow-ops

README.md

@@ -160,6 +160,29 @@ server. Query routing and synchronous pre-accept update validator execution are
 still server-side follow-ups; use those paths only with deployments that
 advertise support for the target workflow type.
 
+Malformed signal and query payloads are reported as typed client errors with
+the server's documented reason and status preserved:
+
+```python
+from durable_workflow import Client, QueryFailed, SignalFailed
+
+client = Client("http://localhost:8080")
+
+try:
+    await client.signal_workflow("counter-1", "increment", args=["not-an-int"])
+except SignalFailed as exc:
+    assert exc.reason == "invalid_signal_arguments"
+    assert exc.status == 422
+    assert exc.validation_errors is not None
+
+try:
+    await client.query_workflow("counter-1", "current-at", args=["not-an-int"])
+except QueryFailed as exc:
+    assert exc.reason == "invalid_query_arguments"
+    assert exc.status == 422
+    assert exc.validation_errors is not None
+```
+
 Use `yield ctx.wait_condition(lambda: self.approved, key="approved",
 timeout=30)` to wait for signal- or update-mutated workflow state without
 polling timers by hand. The SDK sends a stable predicate fingerprint with the

src/durable_workflow/__init__.py

@@ -63,6 +63,7 @@
     ScheduleAlreadyExists,
     ScheduleNotFound,
     ServerError,
+    SignalFailed,
     Unauthorized,
     UpdateRejected,
     WorkflowAlreadyStarted,
@@ -280,6 +281,7 @@
     "RetryPolicy",
     "S3ExternalStorage",
     "ServerError",
+    "SignalFailed",
     "TransportRetryPolicy",
     "Unauthorized",
     "UpdateRejected",

src/durable_workflow/errors.py

@@ -122,11 +122,54 @@ def __init__(self, schedule_id: str) -> None:
 class QueryFailed(DurableWorkflowError):
     """A workflow query was rejected or the workflow raised while handling it."""
 
-    def __init__(self, message: str, *, reason: str | None = None, body: object | None = None) -> None:
+    def __init__(
+        self,
+        message: str,
+        *,
+        reason: str | None = None,
+        status: int | None = None,
+        body: object | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.reason = reason
+        self.status = status
+        self.body = body
+
+    @property
+    def validation_errors(self) -> dict[str, Any] | None:
+        """Return structured query argument validation errors, if the server provided them."""
+        if isinstance(self.body, dict):
+            errors = self.body.get("validation_errors") or self.body.get("errors")
+            if isinstance(errors, dict):
+                return errors
+        return None
+
+
+class SignalFailed(DurableWorkflowError):
+    """A workflow signal was rejected before it could be delivered."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        reason: str | None = None,
+        status: int | None = None,
+        body: object | None = None,
+    ) -> None:
         super().__init__(message)
         self.reason = reason
+        self.status = status
         self.body = body
 
+    @property
+    def validation_errors(self) -> dict[str, Any] | None:
+        """Return structured signal argument validation errors, if the server provided them."""
+        if isinstance(self.body, dict):
+            errors = self.body.get("validation_errors") or self.body.get("errors")
+            if isinstance(errors, dict):
+                return errors
+        return None
+
 
 class WorkflowPayloadDecodeError(DurableWorkflowError):
     """A committed workflow history payload could not be decoded during replay."""
@@ -282,14 +325,30 @@ def _raise_for_status(status: int, body: object, *, context: str = "") -> None:
 
     reason = body.get("reason") if isinstance(body, dict) else None
     message = body.get("message", "") if isinstance(body, dict) else str(body)
+    operation = _control_plane_operation(body)
 
     if status == 401:
         raise Unauthorized(message or "unauthorized")
 
     def query_failed(default: str) -> QueryFailed:
-        return QueryFailed(message or default, reason=reason if isinstance(reason, str) else None, body=body)
+        return QueryFailed(
+            message or default,
+            reason=reason if isinstance(reason, str) else None,
+            status=status,
+            body=body,
+        )
+
+    def signal_failed(default: str) -> SignalFailed:
+        return SignalFailed(
+            message or default,
+            reason=reason if isinstance(reason, str) else None,
+            status=status,
+            body=body,
+        )
 
     if status == 404:
+        if reason == "unknown_signal":
+            raise signal_failed("signal not found")
         if reason in ("query_not_found", "rejected_unknown_query"):
             raise query_failed("query not found")
         if reason == "schedule_not_found":
@@ -305,6 +364,11 @@ def query_failed(default: str) -> QueryFailed:
             raise ScheduleAlreadyExists(context)
         if reason == "duplicate_not_allowed":
             raise WorkflowAlreadyStarted(context)
+        if reason == "run_not_active":
+            if operation == "signal":
+                raise signal_failed("signal rejected")
+            if operation == "query":
+                raise query_failed("query rejected")
         if reason in (
             "query_rejected",
             "query_worker_unavailable",
@@ -326,9 +390,30 @@ def query_failed(default: str) -> QueryFailed:
         raise ServerError(status, body)
 
     if status == 422:
+        if reason == "invalid_signal_arguments":
+            raise signal_failed("signal argument validation failed")
+        if reason == "invalid_query_arguments":
+            raise query_failed("query argument validation failed")
         errors = None
         if isinstance(body, dict):
             errors = body.get("errors") or body.get("validation_errors")
         raise InvalidArgument(message, errors)
 
     raise ServerError(status, body)
+
+
+def _control_plane_operation(body: object) -> str | None:
+    if not isinstance(body, dict):
+        return None
+
+    control_plane = body.get("control_plane")
+    if isinstance(control_plane, dict):
+        operation = control_plane.get("operation")
+        if isinstance(operation, str) and operation:
+            return operation
+
+    operation = body.get("control_plane_operation")
+    if isinstance(operation, str) and operation:
+        return operation
+
+    return None

tests/test_client.py

@@ -23,6 +23,7 @@
     InvalidArgument,
     QueryFailed,
     ServerError,
+    SignalFailed,
     Unauthorized,
     UpdateRejected,
     WorkflowAlreadyStarted,
@@ -573,6 +574,36 @@ async def test_signal_request_matches_polyglot_fixture(self, client: Client) ->
         assert sdk["args"]["workflow_id"] == semantic["workflow_id"]
         assert sdk["args"]["signal_name"] == semantic["signal_name"]
 
+    @pytest.mark.asyncio
+    async def test_unknown_signal_raises_signal_failed(self, client: Client) -> None:
+        resp = _mock_response(404, {"reason": "unknown_signal", "message": "signal [missing] not declared"})
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            pytest.raises(SignalFailed) as excinfo,
+        ):
+            await client.signal_workflow("wf-1", "missing")
+
+        assert excinfo.value.reason == "unknown_signal"
+        assert excinfo.value.status == 404
+
+    @pytest.mark.asyncio
+    async def test_malformed_signal_payload_raises_signal_failed(self, client: Client) -> None:
+        body = {
+            "reason": "invalid_signal_arguments",
+            "message": "signal argument validation failed",
+            "validation_errors": {"n": ["The n argument must be an integer."]},
+        }
+        resp = _mock_response(422, body)
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            pytest.raises(SignalFailed) as excinfo,
+        ):
+            await client.signal_workflow("wf-1", "increment", args=["not-an-int"])
+
+        assert excinfo.value.reason == "invalid_signal_arguments"
+        assert excinfo.value.status == 422
+        assert excinfo.value.validation_errors == {"n": ["The n argument must be an integer."]}
+
 
 class TestWebhookBridgeAdapters:
     @pytest.mark.asyncio
@@ -828,6 +859,25 @@ async def test_worker_routed_unknown_query_raises_query_failed(self, client: Cli
         ):
             await client.query_workflow("wf-1", "status")
 
+    @pytest.mark.asyncio
+    async def test_malformed_query_payload_raises_query_failed(self, client: Client) -> None:
+        body = {
+            "reason": "invalid_query_arguments",
+            "message": "query argument validation failed",
+            "validation_errors": {"n": ["The n argument must be an integer."]},
+        }
+        resp = _mock_response(422, body)
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            pytest.raises(QueryFailed) as excinfo,
+        ):
+            await client.query_workflow("wf-1", "current", args=["not-an-int"])
+
+        assert excinfo.value.reason == "invalid_query_arguments"
+        assert excinfo.value.status == 422
+        assert excinfo.value.body == body
+        assert excinfo.value.validation_errors == {"n": ["The n argument must be an integer."]}
+
     @pytest.mark.asyncio
     async def test_query_rejected(self, client: Client) -> None:
         resp = _mock_response(409, {"reason": "query_rejected", "message": "workflow unavailable"})

tests/test_errors.py

@@ -9,6 +9,7 @@
     NamespaceNotFound,
     QueryFailed,
     ServerError,
+    SignalFailed,
     Unauthorized,
     UpdateRejected,
     WorkflowAlreadyStarted,
@@ -38,6 +39,12 @@ def test_404_query_not_found(self) -> None:
         with pytest.raises(QueryFailed):
             _raise_for_status(404, {"reason": "query_not_found", "message": "query [status] not declared"})
 
+    def test_404_unknown_signal(self) -> None:
+        with pytest.raises(SignalFailed) as exc_info:
+            _raise_for_status(404, {"reason": "unknown_signal", "message": "signal [advance] not declared"})
+        assert exc_info.value.reason == "unknown_signal"
+        assert exc_info.value.status == 404
+
     def test_404_namespace(self) -> None:
         with pytest.raises(NamespaceNotFound):
             _raise_for_status(404, {"reason": "namespace_not_found", "message": "ns missing"})
@@ -54,6 +61,32 @@ def test_409_query_rejected(self) -> None:
         with pytest.raises(QueryFailed):
             _raise_for_status(409, {"reason": "query_rejected", "message": "workflow unavailable"})
 
+    def test_409_terminal_signal_rejected(self) -> None:
+        with pytest.raises(SignalFailed) as exc_info:
+            _raise_for_status(
+                409,
+                {
+                    "reason": "run_not_active",
+                    "message": "workflow is completed",
+                    "control_plane": {"operation": "signal"},
+                },
+            )
+        assert exc_info.value.reason == "run_not_active"
+        assert exc_info.value.status == 409
+
+    def test_409_terminal_query_rejected(self) -> None:
+        with pytest.raises(QueryFailed) as exc_info:
+            _raise_for_status(
+                409,
+                {
+                    "reason": "run_not_active",
+                    "message": "workflow is completed",
+                    "control_plane": {"operation": "query"},
+                },
+            )
+        assert exc_info.value.reason == "run_not_active"
+        assert exc_info.value.status == 409
+
     def test_409_update_rejected(self) -> None:
         with pytest.raises(UpdateRejected):
             _raise_for_status(409, {"reason": "update_rejected", "message": "rejected by handler"})
@@ -67,6 +100,32 @@ def test_422_invalid(self) -> None:
             _raise_for_status(422, {"message": "bad", "errors": {"f": ["req"]}})
         assert exc_info.value.errors == {"f": ["req"]}
 
+    def test_422_invalid_signal_arguments(self) -> None:
+        body = {
+            "reason": "invalid_signal_arguments",
+            "message": "signal argument validation failed",
+            "validation_errors": {"name": ["The name argument must be a string."]},
+        }
+        with pytest.raises(SignalFailed) as exc_info:
+            _raise_for_status(422, body)
+        assert exc_info.value.reason == "invalid_signal_arguments"
+        assert exc_info.value.status == 422
+        assert exc_info.value.body == body
+        assert exc_info.value.validation_errors == {"name": ["The name argument must be a string."]}
+
+    def test_422_invalid_query_arguments(self) -> None:
+        body = {
+            "reason": "invalid_query_arguments",
+            "message": "query argument validation failed",
+            "validation_errors": {"prefix": ["The prefix argument is required."]},
+        }
+        with pytest.raises(QueryFailed) as exc_info:
+            _raise_for_status(422, body)
+        assert exc_info.value.reason == "invalid_query_arguments"
+        assert exc_info.value.status == 422
+        assert exc_info.value.body == body
+        assert exc_info.value.validation_errors == {"prefix": ["The prefix argument is required."]}
+
     def test_500_server_error(self) -> None:
         with pytest.raises(ServerError) as exc_info:
             _raise_for_status(500, {"message": "internal"})