feat(errors): typed per-operation error hierarchy

Author: ayushiahjolia

packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_failure.py

@@ -23,5 +23,5 @@ def test_wait_for_callback_failure(durable_runner):
     assert isinstance(result.error, ErrorObject)
     assert result.error.to_dict() == {
         "ErrorMessage": "my callback error",
-        "ErrorType": "CallableRuntimeError",
+        "ErrorType": "ChildContextError",
     }

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py

@@ -18,9 +18,14 @@
 
 # Most common exceptions - users need to handle these exceptions
 from aws_durable_execution_sdk_python.exceptions import (
+    ChildContextError,
     DurableExecutionsError,
+    DurableOperationError,
     InvocationError,
+    InvokeError,
+    StepError,
     ValidationError,
+    WaitForConditionError,
 )
 
 # Core decorator - used in every durable function
@@ -33,12 +38,17 @@
 
 __all__ = [
     "BatchResult",
+    "ChildContextError",
     "DurableContext",
     "DurableExecutionsError",
+    "DurableOperationError",
     "InvocationError",
+    "InvokeError",
     "ParallelBranch",
     "StepContext",
+    "StepError",
     "ValidationError",
+    "WaitForConditionError",
     "WithRetryConfig",
     "__version__",
     "durable_execution",

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py

@@ -261,7 +261,7 @@ def throw_if_error(self) -> None:
             None,
         )
         if first_error:
-            raise first_error.to_callable_runtime_error()
+            raise first_error.to_durable_operation_error()
 
     def get_results(self) -> list[R]:
         return [

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 import time
-from dataclasses import dataclass
 from enum import Enum
 from typing import TYPE_CHECKING, Self, TypedDict
 
@@ -30,6 +29,11 @@
     }
 )
 
+# Registry mapping a DurableOperationError subclass name to its class, used to
+# reconstruct the correct typed error from a checkpointed ErrorObject on replay.
+# Populated automatically via DurableOperationError.__init_subclass__.
+_DURABLE_OPERATION_ERROR_REGISTRY: dict[str, type[DurableOperationError]] = {}
+
 if TYPE_CHECKING:
     import datetime
 
@@ -266,25 +270,79 @@ class InvalidStateError(DurableExecutionsError):
     """Raised when an operation is attempted on an object in an invalid state."""
 
 
-class UserlandError(DurableExecutionsError):
-    """Failure in user-land - i.e code passed into durable executions from the caller."""
+class DurableOperationError(DurableExecutionsError):
+    """Base class for typed, per-operation failures.
 
+    Wraps a failure that escaped a Durable Function operation (step, invoke,
+    child context, wait_for_condition). The original exception is preserved as
+    ``__cause__`` on the first run; on replay the error is reconstructed from
+    its checkpointed ``ErrorObject`` via :meth:`from_error_fields`.
 
-class CallableRuntimeError(UserlandError):
-    """This error wraps any failure from inside the callable code that you pass to a Durable Function operation."""
+    Attributes:
+        message: Human-readable failure message.
+        error_type: Discriminator identifying the concrete subclass. Defaults
+            to the class name and is serialized as the checkpoint ``ErrorType``.
+        data: Optional serialized error payload, preserved across operation
+            boundaries.
+        stack_trace: Optional stack trace lines captured from the origin.
+    """
 
     def __init__(
         self,
-        message: str | None,
+        message: str | None = None,
+        error_type: str | None = None,
+        data: str | None = None,
+        stack_trace: list[str] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.message: str | None = message
+        self.error_type: str = error_type or type(self).__name__
+        self.data: str | None = data
+        self.stack_trace: list[str] | None = stack_trace
+
+    def __init_subclass__(cls, **kwargs: object) -> None:
+        super().__init_subclass__(**kwargs)
+        _DURABLE_OPERATION_ERROR_REGISTRY[cls.__name__] = cls
+
+    @classmethod
+    def from_error_fields(
+        cls,
         error_type: str | None,
+        message: str | None,
         data: str | None,
         stack_trace: list[str] | None,
-    ) -> None:
-        super().__init__(message)
-        self.message = message
-        self.error_type = error_type
-        self.data = data
-        self.stack_trace = stack_trace
+    ) -> DurableOperationError:
+        """Rebuild the correct subclass from serialized checkpoint error fields.
+
+        Looks the discriminator up in the reconstruction registry, falling back
+        to the base :class:`DurableOperationError` when ``error_type`` is unknown
+        (e.g. a downstream error surfaced by an async invoke/callback checkpoint).
+        """
+        target_cls: type[DurableOperationError] = _DURABLE_OPERATION_ERROR_REGISTRY.get(
+            error_type or "", DurableOperationError
+        )
+        return target_cls(
+            message=message,
+            error_type=error_type,
+            data=data,
+            stack_trace=stack_trace,
+        )
+
+
+class StepError(DurableOperationError):
+    """Raised when a step operation fails."""
+
+
+class InvokeError(DurableOperationError):
+    """Raised when a durable invoke operation fails."""
+
+
+class ChildContextError(DurableOperationError):
+    """Raised when a child context (run_in_child_context, map, parallel) fails."""
+
+
+class WaitForConditionError(DurableOperationError):
+    """Raised when a wait_for_condition operation fails."""
 
 
 class StepInterruptedError(InvocationError):
@@ -397,37 +455,6 @@ def __init__(self, message: str, source_exception: Exception | None = None) -> N
         self.source_exception: Exception | None = source_exception
 
 
-@dataclass(frozen=True)
-class CallableRuntimeErrorSerializableDetails:
-    """Serializable error details."""
-
-    type: str
-    message: str
-
-    @classmethod
-    def from_exception(
-        cls, exception: Exception
-    ) -> CallableRuntimeErrorSerializableDetails:
-        """Create an instance from an Exception, using its type and message.
-
-        Args:
-            exception: An Exception instance
-
-        Returns:
-            A CallableRuntimeErrorDetails instance with the exception's type name and message
-        """
-        return cls(type=exception.__class__.__name__, message=str(exception))
-
-    def __str__(self) -> str:
-        """
-        Return a string representation of the object.
-
-        Returns:
-            A string in the format "type: message"
-        """
-        return f"{self.type}: {self.message}"
-
-
 class SerDesError(DurableExecutionsError):
     """Raised when serialization fails."""
 

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py

@@ -13,8 +13,8 @@
 
 from aws_durable_execution_sdk_python.__about__ import __version__
 from aws_durable_execution_sdk_python.exceptions import (
-    CallableRuntimeError,
     CheckpointError,
+    DurableOperationError,
     GetExecutionStateError,
 )
 
@@ -235,13 +235,41 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> ErrorObject:
 
     @classmethod
     def from_exception(cls, exception: Exception) -> ErrorObject:
+        # Preserve a DurableOperationError's discriminator/data/stack_trace so the
+        # typed info survives across operation boundaries.
+        if isinstance(exception, DurableOperationError):
+            return cls(
+                message=exception.message,
+                type=exception.error_type,
+                data=cls._resolve_error_data(exception),
+                stack_trace=exception.stack_trace,
+            )
         return cls(
             message=str(exception),
             type=type(exception).__name__,
             data=None,
             stack_trace=None,
         )
 
+    @staticmethod
+    def _resolve_error_data(error: DurableOperationError) -> str | None:
+        """Surface the first non-None ``data`` along the cause chain.
+
+        Ensures error data is not lost when a DurableOperationError is re-wrapped
+        across nested operation boundaries (e.g. a step failure escaping through
+        multiple run_in_child_context calls).
+        """
+        if error.data is not None:
+            return error.data
+        node: BaseException | None = error.__cause__
+        for _ in range(10):
+            if node is None:
+                break
+            if isinstance(node, DurableOperationError) and node.data is not None:
+                return node.data
+            node = node.__cause__
+        return None
+
     @classmethod
     def from_message(cls, message: str) -> ErrorObject:
         return cls(
@@ -263,10 +291,10 @@ def to_dict(self) -> MutableMapping[str, Any]:
             result["StackTrace"] = self.stack_trace
         return result
 
-    def to_callable_runtime_error(self) -> CallableRuntimeError:
-        return CallableRuntimeError(
-            message=self.message,
+    def to_durable_operation_error(self) -> DurableOperationError:
+        return DurableOperationError.from_error_fields(
             error_type=self.type,
+            message=self.message,
             data=self.data,
             stack_trace=self.stack_trace,
         )

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py

@@ -7,6 +7,7 @@
 
 from aws_durable_execution_sdk_python.config import ChildConfig
 from aws_durable_execution_sdk_python.exceptions import (
+    ChildContextError,
     InvocationError,
     SuspendExecution,
 )
@@ -78,7 +79,7 @@ def check_result_status(self) -> CheckResult[T]:
             CheckResult indicating the next action to take
 
         Raises:
-            CallableRuntimeError: For FAILED operations
+            ChildContextError: For FAILED operations
         """
         checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result(
             self.operation_identifier.operation_id
@@ -114,7 +115,7 @@ def check_result_status(self) -> CheckResult[T]:
 
         # Terminal failure
         if checkpointed_result.is_failed():
-            checkpointed_result.raise_callable_error()
+            checkpointed_result.raise_operation_error(ChildContextError)
 
         # Create START checkpoint if not exists
         if not checkpointed_result.is_existent() and not self.is_virtual:
@@ -145,7 +146,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T:
         Raises:
             SuspendExecution: Re-raised without checkpointing
             InvocationError: Re-raised after checkpointing FAIL
-            CallableRuntimeError: Raised for other exceptions after checkpointing FAIL
+            ChildContextError: Raised for other exceptions after checkpointing FAIL
         """
         logger.debug(
             "▶️ Executing child context for id: %s, name: %s",
@@ -239,7 +240,24 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T:
             # Don't checkpoint SuspendExecution - let it bubble up
             raise
         except Exception as e:
-            error_object = ErrorObject.from_exception(e)
+            # InvocationError and its derivatives bubble up (with their own type,
+            # not wrapped) so the top-level handler can make the backend retry.
+            if isinstance(e, InvocationError):
+                if not self.is_virtual:
+                    self.state.create_checkpoint(
+                        operation_update=OperationUpdate.create_context_fail(
+                            identifier=self.operation_identifier,
+                            error=ErrorObject.from_exception(e),
+                            sub_type=self.sub_type,
+                        )
+                    )
+                raise
+
+            # Wrap the failure in a ChildContextError (original kept as __cause__)
+            # so the checkpoint records the "ChildContextError" discriminator.
+            child_error: ChildContextError = ChildContextError(message=str(e))
+            child_error.__cause__ = e
+            error_object = ErrorObject.from_exception(child_error)
             # Virtual deliberately does not write checkpoints, but exception still propagates below
             if not self.is_virtual:
                 fail_operation: OperationUpdate = OperationUpdate.create_context_fail(
@@ -252,14 +270,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T:
                 # This guarantees the error is durable and child operations won't be re-executed on replay.
                 self.state.create_checkpoint(operation_update=fail_operation)
 
-            # InvocationError and its derivatives can be retried.
-            # When we encounter an invocation error (in all of its forms), we
-            # bubble that error upwards (with the checkpoint in place for
-            # non-virtual) such that we reach the execution handler at the
-            # very top, which will then make the backend retry.
-            if isinstance(e, InvocationError):
-                raise
-            raise error_object.to_callable_runtime_error() from e
+            raise child_error from e
 
 
 def child_handler(

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py

@@ -5,7 +5,7 @@
 import logging
 from typing import TYPE_CHECKING, TypeVar
 
-from aws_durable_execution_sdk_python.exceptions import ExecutionError
+from aws_durable_execution_sdk_python.exceptions import ExecutionError, InvokeError
 from aws_durable_execution_sdk_python.lambda_service import (
     ChainedInvokeOptions,
     OperationUpdate,
@@ -81,7 +81,7 @@ def check_result_status(self) -> CheckResult[R]:
             CheckResult indicating the next action to take
 
         Raises:
-            CallableRuntimeError: For FAILED, TIMED_OUT, or STOPPED operations
+            InvokeError: For FAILED, TIMED_OUT, or STOPPED operations
             SuspendExecution: For STARTED operations waiting for completion
         """
         checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result(
@@ -107,7 +107,7 @@ def check_result_status(self) -> CheckResult[R]:
             or checkpointed_result.is_timed_out()
             or checkpointed_result.is_stopped()
         ):
-            checkpointed_result.raise_callable_error()
+            checkpointed_result.raise_operation_error(InvokeError)
 
         # Still running - ready to suspend
         if checkpointed_result.is_started():