Add custom FailureDetails properties

CHANGELOG.md

@@ -9,6 +9,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ADDED
 
+- Added `ExceptionPropertiesProvider` and the
+  `exception_properties_provider` worker option to attach portable custom
+  properties to activity, entity, and orchestration failures. Failure
+  properties and recursive inner failures are available through
+  `FailureDetails`, task-failure exceptions, orchestration state, and history.
 - Added the optional `new_version` argument to
 `OrchestrationContext.continue_as_new()` so continued orchestrations can
 switch to a new version.

README.md

@@ -39,6 +39,34 @@ pip install durabletask[azure-blob-payloads]
 See the [feature documentation](./docs/features.md#large-payload-externalization) and the
 [example](./examples/large_payload/) for usage details.
 
+### Failure Details Properties
+
+Configure an `ExceptionPropertiesProvider` to attach portable diagnostic
+properties to activity, entity, and orchestration failures:
+
+```python
+from durabletask import ExceptionPropertiesProvider
+from durabletask.worker import TaskHubGrpcWorker
+
+
+class FailureProperties(ExceptionPropertiesProvider):
+    def get_exception_properties(self, exception: Exception):
+        return {"error_code": getattr(exception, "error_code", None)}
+
+
+worker = TaskHubGrpcWorker(exception_properties_provider=FailureProperties())
+```
+
+Properties support `None`, booleans, numbers, strings, nested mappings, and
+lists. They are available from `TaskFailedError.details`, orchestration state,
+and history as `FailureDetails.properties`.
+
+> [!NOTE]
+> Python intentionally treats `.NET`'s `dt:` and `dto:` property strings as
+> ordinary strings, matching the Java SDK. Consequently, `datetime` values do
+> not retain their type across Python/.NET failure-details property round trips;
+> this is a .NET SDK parity gap.
+
 ## Trademarks
 
 This project may contain trademarks or logos for projects, products, or services. Authorized use of

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added `exception_properties_provider` to `DurableTaskSchedulerWorker` for
+  attaching portable custom properties to activity, entity, and orchestration
+  failures.
+
 ## v1.9.0
 
 CHANGED

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -16,10 +16,14 @@
     GrpcChannelOptions,
     GrpcWorkerResiliencyOptions,
 )
+from durabletask.exception_properties import ExceptionPropertiesProvider
 import durabletask.internal.shared as shared
 from durabletask.payload.store import PayloadStore
 from durabletask.serialization import DataConverter
-from durabletask.worker import ConcurrencyOptions, TaskHubGrpcWorker
+from durabletask.worker import (
+    ConcurrencyOptions,
+    TaskHubGrpcWorker,
+)
 
 
 # Worker class used for Durable Task Scheduler (DTS)
@@ -46,6 +50,8 @@ class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
             externalizing large payloads. If None, payloads are sent inline.
         log_handler (logging.Handler | None, optional): Custom logging handler for worker logs.
         log_formatter (logging.Formatter | None, optional): Custom log formatter for worker logs.
+        exception_properties_provider (ExceptionPropertiesProvider | None, optional):
+            Extracts portable custom properties from exceptions reported by this worker.
 
     Raises:
         ValueError: If taskhub is empty or None.
@@ -84,7 +90,8 @@ def __init__(self, *,
                  payload_store: PayloadStore | None = None,
                  data_converter: DataConverter | None = None,
                  log_handler: logging.Handler | None = None,
-                 log_formatter: logging.Formatter | None = None):
+                 log_formatter: logging.Formatter | None = None,
+                 exception_properties_provider: ExceptionPropertiesProvider | None = None):
 
         if not taskhub:
             raise ValueError("The taskhub value cannot be empty.")
@@ -113,5 +120,6 @@ def __init__(self, *,
             # DTS natively supports long timers so chunking is unnecessary
             maximum_timer_interval=None,
             payload_store=payload_store,
-            data_converter=data_converter
+            data_converter=data_converter,
+            exception_properties_provider=exception_properties_provider,
         )

durabletask/__init__.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
+    from durabletask.exception_properties import ExceptionPropertiesProvider
     from durabletask.grpc_options import GrpcChannelOptions, GrpcRetryPolicyOptions
     from durabletask.payload.store import LargePayloadStorageOptions, PayloadStore
     from durabletask.worker import (
@@ -21,6 +22,7 @@
     "ActivityWorkItemFilter",
     "ConcurrencyOptions",
     "EntityWorkItemFilter",
+    "ExceptionPropertiesProvider",
     "GrpcChannelOptions",
     "GrpcRetryPolicyOptions",
     "LargePayloadStorageOptions",
@@ -42,6 +44,7 @@
     "ActivityWorkItemFilter": "durabletask.worker",
     "ConcurrencyOptions": "durabletask.worker",
     "EntityWorkItemFilter": "durabletask.worker",
+    "ExceptionPropertiesProvider": "durabletask.exception_properties",
     "GrpcChannelOptions": "durabletask.grpc_options",
     "GrpcRetryPolicyOptions": "durabletask.grpc_options",
     "LargePayloadStorageOptions": "durabletask.payload.store",

durabletask/client.py

@@ -254,10 +254,7 @@ def parse_orchestration_state(
         data_converter: DataConverter | None = None) -> OrchestrationState:
     failure_details = None
     if state.failureDetails.errorMessage != '' or state.failureDetails.errorType != '':
-        failure_details = task.FailureDetails(
-            state.failureDetails.errorMessage,
-            state.failureDetails.errorType,
-            state.failureDetails.stackTrace.value if not helpers.is_empty(state.failureDetails.stackTrace) else None)
+        failure_details = helpers.failure_details_from_protobuf(state.failureDetails)
 
     return OrchestrationState(
         state.instanceId,

durabletask/exception_properties.py

@@ -0,0 +1,15 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Types for enriching task failure details."""
+
+from collections.abc import Mapping
+from typing import Any, Protocol
+
+
+class ExceptionPropertiesProvider(Protocol):
+    """Extract portable custom properties from an exception."""
+
+    def get_exception_properties(self, exception: Exception) -> Mapping[str, Any] | None:
+        """Return properties to include in the exception's failure details."""
+        ...

durabletask/extensions/history_export/serialization.py

@@ -52,6 +52,20 @@ def event_to_dict(event: history.HistoryEvent) -> dict[str, Any]:
     return {"event_type": type(event).__name__, **payload}
 
 
+def failure_details_to_dict(failure: task.FailureDetails) -> dict[str, Any]:
+    """Convert recursively nested failure details into a JSON-safe dict."""
+    return {
+        "message": failure.message,
+        "error_type": failure.error_type,
+        "stack_trace": failure.stack_trace,
+        "inner_failure": (
+            failure_details_to_dict(failure.inner_failure)
+            if failure.inner_failure is not None else None
+        ),
+        "properties": failure.properties,
+    }
+
+
 def orchestration_state_to_dict(
     state: client_module.OrchestrationState,
 ) -> dict[str, Any]:
@@ -61,20 +75,7 @@ def orchestration_state_to_dict(
     class names or module paths appear in the resulting dict.
     """
     failure = state.failure_details
-    failure_dict: dict[str, Any] | None = None
-    if failure is not None:
-        failure_dict = {
-            "message": failure.message,
-            "error_type": failure.error_type,
-            "stack_trace": failure.stack_trace,
-        }
-        inner = getattr(failure, "inner_failure", None)
-        if isinstance(inner, task.FailureDetails):
-            failure_dict["inner_failure"] = {
-                "message": inner.message,
-                "error_type": inner.error_type,
-                "stack_trace": inner.stack_trace,
-            }
+    failure_dict = failure_details_to_dict(failure) if failure is not None else None
     return {
         "instance_id": state.instance_id,
         "name": state.name,

durabletask/history.py

@@ -13,6 +13,7 @@
 from google.protobuf.message import Message
 
 from durabletask import task
+import durabletask.internal.helpers as helpers
 import durabletask.internal.orchestrator_service_pb2 as pb
 
 
@@ -279,11 +280,7 @@ def _failure_details(msg: Message, field_name: str) -> task.FailureDetails | Non
     if not msg.HasField(field_name):
         return None
     details = getattr(msg, field_name)
-    return task.FailureDetails(
-        details.errorMessage,
-        details.errorType,
-        details.stackTrace.value if details.HasField('stackTrace') else None,
-    )
+    return helpers.failure_details_from_protobuf(details)
 
 
 def _trace_context(msg: Message, field_name: str) -> TraceContext | None:

durabletask/internal/helpers.py

@@ -1,15 +1,22 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import logging
 import traceback
+from collections.abc import Mapping, Sequence
+from decimal import Decimal
 from datetime import datetime, timezone
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
-from google.protobuf import timestamp_pb2, wrappers_pb2
+from google.protobuf import struct_pb2, timestamp_pb2, wrappers_pb2
 
 from durabletask.entities import EntityInstanceId
+from durabletask.exception_properties import ExceptionPropertiesProvider
 import durabletask.internal.orchestrator_service_pb2 as pb
 
+if TYPE_CHECKING:
+    from durabletask.task import FailureDetails
+
 # TODO: The new_xxx_event methods are only used by test code and should be moved elsewhere
 
 
@@ -141,32 +148,143 @@ def get_qualified_name(t: type) -> str:
     return f"{module}.{qualname}"
 
 
-def new_failure_details(ex: Exception, _visited: set[int] | None = None) -> pb.TaskFailureDetails:
+def protobuf_value_from_python(value: Any) -> struct_pb2.Value:
+    """Convert a portable Python value to a protobuf ``Value``."""
+    if value is None:
+        return struct_pb2.Value(null_value=struct_pb2.NullValue.NULL_VALUE)
+    if isinstance(value, bool):
+        return struct_pb2.Value(bool_value=value)
+    if isinstance(value, (int, float, Decimal)):
+        return struct_pb2.Value(number_value=float(value))
+    if isinstance(value, str):
+        return struct_pb2.Value(string_value=value)
+    if isinstance(value, Mapping):
+        mapping = cast(Mapping[Any, Any], value)
+        fields = {
+            str(key): protobuf_value_from_python(item)
+            for key, item in mapping.items()
+        }
+        return struct_pb2.Value(struct_value=struct_pb2.Struct(fields=fields))
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        sequence = cast(Sequence[Any], value)
+        return struct_pb2.Value(
+            list_value=struct_pb2.ListValue(
+                values=[protobuf_value_from_python(item) for item in sequence]))
+    return struct_pb2.Value(string_value=str(value))
+
+
+def python_value_from_protobuf(value: struct_pb2.Value) -> Any:
+    """Convert a protobuf ``Value`` to a JSON-safe Python value."""
+    kind = value.WhichOneof("kind")
+    if kind == "null_value":
+        return None
+    if kind == "bool_value":
+        return value.bool_value
+    if kind == "number_value":
+        return value.number_value
+    if kind == "string_value":
+        return value.string_value
+    if kind == "struct_value":
+        return {
+            key: python_value_from_protobuf(item)
+            for key, item in value.struct_value.fields.items()
+        }
+    if kind == "list_value":
+        return [python_value_from_protobuf(item) for item in value.list_value.values]
+    return str(value)
+
+
+def failure_details_from_protobuf(details: pb.TaskFailureDetails) -> "FailureDetails":
+    """Convert protobuf failure details to the public, JSON-safe model."""
+    from durabletask.task import FailureDetails
+
+    return FailureDetails(
+        message=details.errorMessage,
+        error_type=details.errorType,
+        stack_trace=details.stackTrace.value if details.HasField("stackTrace") else None,
+        inner_failure=(
+            failure_details_from_protobuf(details.innerFailure)
+            if details.HasField("innerFailure") else None),
+        properties=(
+            {
+                key: python_value_from_protobuf(value)
+                for key, value in details.properties.items()
+            }
+            if details.properties else None),
+    )
+
+
+def new_failure_details(
+        ex: Exception,
+        exception_properties_provider: ExceptionPropertiesProvider | None = None,
+        logger: logging.Logger | None = None,
+        _visited: set[int] | None = None) -> pb.TaskFailureDetails:
     if _visited is None:
         _visited = set()
     _visited.add(id(ex))
     inner: BaseException | None = ex.__cause__ or ex.__context__
     if len(_visited) > 10 or (inner and id(inner) in _visited) or not isinstance(inner, Exception):
         inner = None
-    return pb.TaskFailureDetails(
+    properties: dict[str, struct_pb2.Value] | None = None
+    if exception_properties_provider is not None:
+        try:
+            provider_properties = exception_properties_provider.get_exception_properties(ex)
+        except Exception:
+            if logger is not None:
+                logger.warning(
+                    "ExceptionPropertiesProvider failed while processing %s.",
+                    get_qualified_name(type(ex)),
+                    exc_info=True)
+        else:
+            try:
+                untyped_properties: object = cast(object, provider_properties)
+                if untyped_properties is not None:
+                    if not isinstance(untyped_properties, Mapping):
+                        raise TypeError(
+                            "ExceptionPropertiesProvider.get_exception_properties() "
+                            "must return a mapping or None.")
+                    properties = {
+                        str(key): protobuf_value_from_python(value)
+                        for key, value in cast(Mapping[Any, Any], untyped_properties).items()
+                    }
+            except Exception:
+                if logger is not None:
+                    logger.warning(
+                        "ExceptionPropertiesProvider returned invalid properties for %s.",
+                        get_qualified_name(type(ex)),
+                        exc_info=True)
+
+    failure_details = pb.TaskFailureDetails(
         errorType=get_qualified_name(type(ex)),
         errorMessage=str(ex),
         stackTrace=wrappers_pb2.StringValue(value=''.join(traceback.format_tb(ex.__traceback__))),
-        innerFailure=new_failure_details(inner, _visited) if inner else None
+        innerFailure=(
+            new_failure_details(
+                inner, exception_properties_provider, logger, _visited)
+            if inner else None)
     )
+    if properties:
+        for key, value in properties.items():
+            failure_details.properties[key].CopyFrom(value)
+    return failure_details
 
 
 def _failure_details_from_core_dict(fd: dict[str, Any]) -> pb.TaskFailureDetails:
     """Convert a serialized DurableTask.Core ``FailureDetails`` dict to protobuf."""
     inner = fd.get("InnerFailure")
     stack_trace = fd.get("StackTrace")
-    return pb.TaskFailureDetails(
+    result = pb.TaskFailureDetails(
         errorType=str(fd.get("ErrorType") or ""),
         errorMessage=str(fd.get("ErrorMessage") or ""),
         stackTrace=get_string_value(str(stack_trace) if stack_trace is not None else None),
         innerFailure=_failure_details_from_core_dict(cast(dict[str, Any], inner)) if isinstance(inner, dict) else None,
         isNonRetriable=bool(fd.get("IsNonRetriable", False)),
     )
+    properties = fd.get("Properties")
+    if isinstance(properties, Mapping):
+        for key, value in cast(Mapping[Any, Any], properties).items():
+            result.properties[str(key)].CopyFrom(protobuf_value_from_python(value))
+    return result
 
 
 def entity_response_failure_details(