feat(errors): typed per-operation error hierarchy

Author: ayushiahjolia

packages/aws-durable-execution-sdk-python-examples/examples-catalog.json

@@ -748,6 +748,17 @@
         "FILESYSTEM_MOUNT_PATH": "/mnt/s3"
       },
       "path": "./src/filesystem_serdes/filesystem_serdes_preview.py"
+    },
+    {
+      "name": "Catch Typed Step Error",
+      "description": "Catch a typed StepError raised by a failed step",
+      "handler": "catch_typed_step_error.handler",
+      "integration": true,
+      "durableConfig": {
+        "RetentionPeriodInDays": 7,
+        "ExecutionTimeout": 300
+      },
+      "path": "./src/error_handling/catch_typed_step_error.py"
     }
   ]
 }

packages/aws-durable-execution-sdk-python-examples/src/error_handling/catch_typed_step_error.py

@@ -0,0 +1,43 @@
+"""Demonstrates catching a typed StepError raised by a failed step.
+
+A failure inside a durable operation surfaces as a typed
+DurableOperationError subclass (StepError, InvokeError, ChildContextError,
+WaitForConditionError) rather than a single catch-all error, so callers can
+handle each operation's failure distinctly.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python import StepError
+from aws_durable_execution_sdk_python.config import StepConfig
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.retries import (
+    RetryStrategyConfig,
+    create_retry_strategy,
+)
+from aws_durable_execution_sdk_python.types import StepContext
+
+
+@durable_execution
+def handler(_event: Any, context: DurableContext) -> dict[str, Any]:
+    """Run a step that fails and catch the typed StepError."""
+
+    def failing_step(_step_context: StepContext) -> None:
+        raise ValueError("payment declined")
+
+    # A single attempt so the step fails without scheduling retries.
+    config = StepConfig(
+        retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1))
+    )
+
+    try:
+        context.step(failing_step, name="charge-card", config=config)
+    except StepError as error:
+        return {
+            "handled": True,
+            "error_type": error.error_type,
+            "message": error.message,
+        }
+
+    return {"handled": False}

packages/aws-durable-execution-sdk-python-examples/test/error_handling/test_catch_typed_step_error.py

@@ -0,0 +1,27 @@
+"""Tests for catch_typed_step_error."""
+
+import pytest
+from aws_durable_execution_sdk_python.execution import InvocationStatus
+
+from src.error_handling import catch_typed_step_error
+from test.conftest import deserialize_operation_payload
+
+
+@pytest.mark.example
+@pytest.mark.durable_execution(
+    handler=catch_typed_step_error.handler,
+    lambda_function_name="Catch Typed Step Error",
+)
+def test_catch_typed_step_error(durable_runner):
+    """A failed step surfaces as a catchable StepError with a typed discriminator."""
+    with durable_runner:
+        result = durable_runner.run(input=None, timeout=10)
+
+    assert result.status is InvocationStatus.SUCCEEDED
+
+    result_data = deserialize_operation_payload(result.result)
+    assert result_data == {
+        "handled": True,
+        "error_type": "StepError",
+        "message": "payment declined",
+    }

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,
         )