Add payload validation error factory

CHANGELOG.md

@@ -20,6 +20,8 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Added `temporalio.converter.create_payload_validation_error` to create the
+  non-retryable application error used when a converted payload fails validation.
 - Added experimental `temporalio.contrib.opentelemetry.ReplaySafeMeterProvider` and
   `ReplaySafeLoggerProvider` (and exported `ReplaySafeTracerProvider`): wrap an
   OpenTelemetry provider so metrics and log events recorded from workflow code (e.g. by

temporalio/converter/__init__.py

@@ -37,6 +37,9 @@
     transfer_type_convertible,
     value_to_type,
 )
+from temporalio.converter._payload_validation_error import (
+    create_payload_validation_error,
+)
 from temporalio.converter._search_attributes import (
     decode_search_attributes,
     decode_typed_search_attributes,
@@ -86,6 +89,7 @@
     "decode_search_attributes",
     "decode_typed_search_attributes",
     "default",
+    "create_payload_validation_error",
     "encode_search_attribute_values",
     "encode_search_attributes",
     "encode_typed_search_attribute_value",

temporalio/converter/_payload_validation_error.py

@@ -0,0 +1,25 @@
+"""Payload validation error helpers."""
+
+from typing import Any
+
+import temporalio.exceptions
+
+
+def create_payload_validation_error(
+    details: Any,
+) -> temporalio.exceptions.ApplicationError:
+    """Create an error indicating that a converted payload failed validation.
+
+    Args:
+        details: Structured details describing the validation failure.
+
+    Returns:
+        A non-retryable application error with the reserved payload validation
+        failure type.
+    """
+    return temporalio.exceptions.ApplicationError(
+        "Payload validation failed",
+        details,
+        type="PayloadValidationError",
+        non_retryable=True,
+    )

tests/nexus/test_workflow_caller_errors.py

@@ -29,7 +29,12 @@
     Client,
     WorkflowFailureError,
 )
-from temporalio.converter import DataConverter, DefaultPayloadConverter, PayloadCodec
+from temporalio.converter import (
+    DataConverter,
+    DefaultPayloadConverter,
+    PayloadCodec,
+    create_payload_validation_error,
+)
 from temporalio.exceptions import (
     ApplicationError,
     NexusOperationError,
@@ -844,7 +849,10 @@ async def test_nexus_operation_fails_without_retry_on_converter_failure(
             pytest.fail("Expected WorkflowFailureError")
 
 
-_PAYLOAD_VALIDATION_FAILURE_MESSAGE = "Nexus operation input failed validation"
+_PAYLOAD_VALIDATION_FAILURE_MESSAGE = "Payload validation failed"
+_PAYLOAD_VALIDATION_FAILURE_DETAILS = {
+    "violations": [{"path": "some.path", "reason": "must be an int"}]
+}
 
 
 class RaiseOnDecodeCodec(PayloadCodec):
@@ -893,10 +901,8 @@ async def _deserialize_input_with_converter_error(error: Exception) -> Any:
 
 
 async def test_codec_input_payload_validation_failure_is_bad_request():
-    validation_error = ApplicationError(
-        _PAYLOAD_VALIDATION_FAILURE_MESSAGE,
-        type="PayloadValidationError",
-        non_retryable=True,
+    validation_error = create_payload_validation_error(
+        _PAYLOAD_VALIDATION_FAILURE_DETAILS
     )
     with pytest.raises(nexusrpc.HandlerError) as err:
         await _deserialize_input_with_codec_error(validation_error)
@@ -942,10 +948,8 @@ async def test_retryable_codec_input_payload_validation_failure_is_internal():
 
 
 async def test_converter_input_payload_validation_failure_is_bad_request():
-    validation_error = ApplicationError(
-        _PAYLOAD_VALIDATION_FAILURE_MESSAGE,
-        type="PayloadValidationError",
-        non_retryable=True,
+    validation_error = create_payload_validation_error(
+        _PAYLOAD_VALIDATION_FAILURE_DETAILS
     )
     with pytest.raises(nexusrpc.HandlerError) as err:
         await _deserialize_input_with_converter_error(validation_error)
@@ -1002,10 +1006,8 @@ async def test_nexus_operation_fails_without_retry_on_codec_input_validation_fai
         pytest.skip("Nexus tests don't work with time-skipping server")
 
     task_queue = str(uuid.uuid4())
-    validation_error = ApplicationError(
-        _PAYLOAD_VALIDATION_FAILURE_MESSAGE,
-        type="PayloadValidationError",
-        non_retryable=True,
+    validation_error = create_payload_validation_error(
+        _PAYLOAD_VALIDATION_FAILURE_DETAILS
     )
     handler_client = Client(
         client.service_client,
@@ -1053,6 +1055,7 @@ async def test_nexus_operation_fails_without_retry_on_codec_input_validation_fai
         assert isinstance(cause, ApplicationError)
         assert cause.type == "PayloadValidationError"
         assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE
+        assert cause.details == (_PAYLOAD_VALIDATION_FAILURE_DETAILS,)
 
 
 async def test_nexus_operation_fails_without_retry_on_converter_input_validation_failure(
@@ -1062,10 +1065,8 @@ async def test_nexus_operation_fails_without_retry_on_converter_input_validation
         pytest.skip("Nexus tests don't work with time-skipping server")
 
     task_queue = str(uuid.uuid4())
-    validation_error = ApplicationError(
-        _PAYLOAD_VALIDATION_FAILURE_MESSAGE,
-        type="PayloadValidationError",
-        non_retryable=True,
+    validation_error = create_payload_validation_error(
+        _PAYLOAD_VALIDATION_FAILURE_DETAILS
     )
     handler_client = Client(
         client.service_client,
@@ -1114,3 +1115,4 @@ async def test_nexus_operation_fails_without_retry_on_converter_input_validation
         assert isinstance(cause, ApplicationError)
         assert cause.type == "PayloadValidationError"
         assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE
+        assert cause.details == (_PAYLOAD_VALIDATION_FAILURE_DETAILS,)

tests/test_converter.py

@@ -49,6 +49,7 @@
     JSONTypeConverterUnhandled,
     PayloadCodec,
     TransferTypeConverter,
+    create_payload_validation_error,
     decode_search_attributes,
     encode_search_attribute_values,
     transfer_type_convertible,
@@ -106,6 +107,25 @@ class NewTypeMessage:
     data: dict[MyNewTypeStr, str]
 
 
+def test_create_payload_validation_error() -> None:
+    details = {"violations": [{"path": "some.path", "reason": "must be an int"}]}
+
+    err = create_payload_validation_error(details)
+
+    assert err.message == "Payload validation failed"
+    assert err.type == "PayloadValidationError"
+    assert err.non_retryable
+    assert err.details == (details,)
+
+    failure = Failure()
+    DataConverter.default.failure_converter.to_failure(
+        err, DataConverter.default.payload_converter, failure
+    )
+    assert DataConverter.default.payload_converter.from_payloads(
+        failure.application_failure_info.details.payloads
+    ) == [details]
+
+
 async def test_converter_default():
     async def assert_payload(
         input,  # type:ignore[reportMissingParameterType]