diff --git a/durabletask/internal/helpers.py b/durabletask/internal/helpers.py
index 0b1f655..0cd2d40 100644
--- a/durabletask/internal/helpers.py
+++ b/durabletask/internal/helpers.py
@@ -116,11 +116,18 @@ def new_sub_orchestration_failed_event(event_id: int, ex: Exception) -> pb.Histo
     )
 
 
-def new_failure_details(ex: Exception) -> pb.TaskFailureDetails:
+def new_failure_details(ex: Exception, _visited: Optional[set[int]] = None) -> pb.TaskFailureDetails:
+    if _visited is None:
+        _visited = set()
+    _visited.add(id(ex))
+    inner: Optional[BaseException] = 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(
         errorType=type(ex).__name__,
         errorMessage=str(ex),
-        stackTrace=wrappers_pb2.StringValue(value=''.join(traceback.format_tb(ex.__traceback__)))
+        stackTrace=wrappers_pb2.StringValue(value=''.join(traceback.format_tb(ex.__traceback__))),
+        innerFailure=new_failure_details(inner, _visited) if inner else None
     )
 
 
diff --git a/durabletask/task.py b/durabletask/task.py
index 1ae9f49..50d0d05 100644
--- a/durabletask/task.py
+++ b/durabletask/task.py
@@ -302,8 +302,10 @@ def stack_trace(self) -> Optional[str]:
 class TaskFailedError(Exception):
     """Exception type for all orchestration task failures."""
 
-    def __init__(self, message: str, details: pb.TaskFailureDetails):
+    def __init__(self, message: str, details: Union[pb.TaskFailureDetails, Exception]):
         super().__init__(message)
+        if isinstance(details, Exception):
+            details = pbh.new_failure_details(details)
         self._details = FailureDetails(
             details.errorMessage,
             details.errorType,
@@ -424,7 +426,7 @@ def complete(self, result: T):
         if self._parent is not None:
             self._parent.on_child_completed(self)
 
-    def fail(self, message: str, details: pb.TaskFailureDetails):
+    def fail(self, message: str, details: Union[Exception, pb.TaskFailureDetails]):
         if self._is_complete:
             raise ValueError('The task has already completed.')
         self._exception = TaskFailedError(message, details)