Preserve nested failure metadata

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 26daf565-a1f4-4bf0-9540-b57f31a4a12b

Author: andystaples

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/task.py

@@ -579,12 +579,18 @@ def __init__(self, message: str, details: pb.TaskFailureDetails | Exception):
                 details.innerFailure.CopyFrom(nested_failure)
                 for key, value in nested_failure.properties.items():
                     details.properties[key].CopyFrom(value)
+        self._failure_details = details
         self._details = pbh.failure_details_from_protobuf(details)
 
     @property
     def details(self) -> FailureDetails:
         return self._details
 
+    @property
+    def failure_details(self) -> pb.TaskFailureDetails:
+        """Return the underlying protobuf failure details."""
+        return self._failure_details
+
 
 class NonDeterminismError(Exception):
     pass

durabletask/worker.py

@@ -1624,12 +1624,21 @@ def set_failed(self, ex: Exception | pb.TaskFailureDetails):
         # self._pending_actions.clear()  # Cancel any pending actions
         self._completion_status = pb.ORCHESTRATION_STATUS_FAILED
 
+        if isinstance(ex, task.TaskFailedError):
+            failure_details = ph.new_failure_details(
+                ex, self._exception_properties_provider, self._logger)
+            failure_details.innerFailure.CopyFrom(ex.failure_details)
+        elif isinstance(ex, Exception):
+            failure_details = ph.new_failure_details(
+                ex, self._exception_properties_provider, self._logger)
+        else:
+            failure_details = ex
+
         action = ph.new_complete_orchestration_action(
             self.next_sequence_number(),
             pb.ORCHESTRATION_STATUS_FAILED,
             None,
-            ph.new_failure_details(
-                ex, self._exception_properties_provider, self._logger) if isinstance(ex, Exception) else ex,
+            failure_details,
         )
         self._pending_actions[action.id] = action
 

tests/durabletask/extensions/history_export/test_serialization.py

@@ -184,13 +184,42 @@ def test_state_to_dict_has_no_python_type_metadata(self) -> None:
     def test_state_with_failure_details(self) -> None:
         st = self._state()
         st.failure_details = task.FailureDetails(
-            message="boom", error_type="RuntimeError", stack_trace="trace",
+            message="boom",
+            error_type="RuntimeError",
+            stack_trace="trace",
+            properties={"code": "OUTER"},
+            inner_failure=task.FailureDetails(
+                message="middle",
+                error_type="ValueError",
+                stack_trace=None,
+                properties={"code": "MIDDLE"},
+                inner_failure=task.FailureDetails(
+                    message="inner",
+                    error_type="KeyError",
+                    stack_trace=None,
+                    properties={"code": "INNER"},
+                ),
+            ),
         )
         d = orchestration_state_to_dict(st)
         assert d["failure_details"] == {
             "message": "boom",
             "error_type": "RuntimeError",
             "stack_trace": "trace",
+            "properties": {"code": "OUTER"},
+            "inner_failure": {
+                "message": "middle",
+                "error_type": "ValueError",
+                "stack_trace": None,
+                "properties": {"code": "MIDDLE"},
+                "inner_failure": {
+                    "message": "inner",
+                    "error_type": "KeyError",
+                    "stack_trace": None,
+                    "properties": {"code": "INNER"},
+                    "inner_failure": None,
+                },
+            },
         }
 
     def test_metadata_embedded_in_json(self) -> None:

tests/durabletask/test_orchestration_executor.py

@@ -718,6 +718,46 @@ def orchestrator(ctx: task.OrchestrationContext, orchestrator_input):
     assert user_code_statement in complete_action.failureDetails.stackTrace.value
 
 
+def test_unhandled_activity_failure_preserves_failure_details():
+    """An unhandled task failure keeps its original metadata as an inner failure."""
+    def dummy_activity(ctx, _):
+        pass
+
+    def orchestrator(ctx: task.OrchestrationContext, orchestrator_input):
+        yield ctx.call_activity(dummy_activity, input=orchestrator_input)
+
+    class PropertiesProvider:
+        def get_exception_properties(self, exception: Exception):
+            if isinstance(exception, ValueError):
+                return {"code": "VALUE"}
+            return None
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+    old_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+        helpers.new_task_scheduled_event(1, task.get_name(dummy_activity)),
+    ]
+    executor = worker._OrchestrationExecutor(
+        registry,
+        TEST_LOGGER,
+        JsonDataConverter(),
+        exception_properties_provider=PropertiesProvider(),
+    )
+    activity_failure = ValueError("boom")
+    failed_event = helpers.new_task_failed_event(1, activity_failure)
+    failed_event.taskFailed.failureDetails.CopyFrom(
+        helpers.new_failure_details(activity_failure, PropertiesProvider()))
+
+    result = executor.execute(TEST_INSTANCE_ID, old_events, [failed_event])
+
+    failure = get_and_validate_complete_orchestration_action_list(1, result.actions).failureDetails
+    assert failure.errorType == "durabletask.task.TaskFailedError"
+    assert failure.innerFailure.errorType == "builtins.ValueError"
+    assert failure.innerFailure.properties["code"].string_value == "VALUE"
+
+
 def test_activity_retry_policies():
     """Tests the retry policy logic for activity tasks"""