feat(errors): typed per-operation error hierarchy

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

@@ -763,6 +763,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,59 @@
+"""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.
+
+The handler also crosses a wait/replay boundary after catching the error. This
+shows the typed StepError is deterministic across replay: on the resume
+invocation the handler replays from the top, the failed step is reconstructed
+from its checkpoint (its body is NOT re-executed), and it surfaces the identical
+StepError, which the same ``except`` catches again.
+"""
+
+from typing import Any
+
+from aws_durable_execution_sdk_python import StepError
+from aws_durable_execution_sdk_python.config import Duration, 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, catch the typed StepError, then replay across a wait."""
+
+    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))
+    )
+
+    outcome: dict[str, Any] = {"handled": False}
+    try:
+        context.step(failing_step, name="charge-card", config=config)
+    except StepError as error:
+        # The class identifies the operation kind (StepError); error_type carries
+        # the original failure type (ValueError).
+        outcome = {
+            "handled": True,
+            "caught": type(error).__name__,
+            "error_type": error.error_type,
+            "message": error.message,
+        }
+
+    # Replay boundary: the execution suspends here and resumes as a new
+    # invocation that replays from the top. On resume, the failed step above is
+    # reconstructed from its checkpoint and raises the same StepError, which the
+    # except catches again — so `outcome` is identical on first run and replay.
+    context.wait(duration=Duration.from_seconds(1), name="post-failure-cooldown")
+
+    return outcome

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

@@ -0,0 +1,62 @@
+"""Tests for catch_typed_step_error."""
+
+import pytest
+from aws_durable_execution_sdk_python.execution import InvocationStatus
+from aws_durable_execution_sdk_python.lambda_service import (
+    OperationStatus,
+    OperationType,
+)
+
+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, identical across replay.
+
+    The handler catches the StepError and then crosses a wait/replay boundary, so
+    the returned ``outcome`` is produced on the resume (replay) invocation, while
+    the STEP FAILED checkpoint was written on the first invocation. Asserting the
+    surfaced error matches the persisted checkpoint error shows the typed error is
+    reconstructed identically across replays.
+    """
+    with durable_runner:
+        result = durable_runner.run(input=None, timeout=10)
+
+    assert result.status is InvocationStatus.SUCCEEDED
+
+    # The error caught on the replay pass (returned as the execution result).
+    result_data = deserialize_operation_payload(result.result)
+    assert result_data == {
+        "handled": True,
+        "caught": "StepError",
+        "error_type": "ValueError",
+        "message": "payment declined",
+    }
+
+    # A WAIT operation proves the execution suspended and replayed after the
+    # error was caught, so the result above reflects the replay invocation.
+    wait_ops = [
+        op for op in result.operations if op.operation_type == OperationType.WAIT
+    ]
+    assert len(wait_ops) >= 1
+
+    # The step's own checkpoint (written on the first run) is FAILED and records
+    # the raw escaping error type.
+    step_ops = [
+        op for op in result.operations if op.operation_type == OperationType.STEP
+    ]
+    assert len(step_ops) == 1
+    checkpoint_error = step_ops[0].error
+    assert step_ops[0].status is OperationStatus.FAILED
+    assert checkpoint_error is not None
+
+    # Cross-replay identity: the error surfaced on the replay pass (result_data)
+    # matches the error persisted on the first run (the checkpoint).
+    assert result_data["error_type"] == checkpoint_error.type == "ValueError"
+    assert result_data["message"] == checkpoint_error.message == "payment declined"

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

@@ -21,7 +21,11 @@ def test_wait_for_callback_failure(durable_runner):
 
     assert result.status is InvocationStatus.FAILED
     assert isinstance(result.error, ErrorObject)
+    # The callback failure raises CallbackError inside the child context that
+    # wait_for_callback runs in; it is wrapped as ChildContextError so first run
+    # and replay surface an identical type. The original CallbackError type is
+    # preserved on the error's error_type/__cause__.
     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,15 @@
 
 # Most common exceptions - users need to handle these exceptions
 from aws_durable_execution_sdk_python.exceptions import (
+    ChildContextError,
     DurableExecutionsError,
+    DurableOperationError,
+    ExecutionError,
     InvocationError,
+    InvokeError,
+    StepError,
     ValidationError,
+    WaitForConditionError,
 )
 
 # Core decorator - used in every durable function
@@ -33,12 +39,18 @@
 
 __all__ = [
     "BatchResult",
+    "ChildContextError",
     "DurableContext",
     "DurableExecutionsError",
+    "DurableOperationError",
+    "ExecutionError",
     "InvocationError",
+    "InvokeError",
     "ParallelBranch",
     "StepContext",
+    "StepError",
     "ValidationError",
+    "WaitForConditionError",
     "WithRetryConfig",
     "__version__",
     "durable_execution",

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

@@ -25,6 +25,7 @@
     NestingType,
 )
 from aws_durable_execution_sdk_python.exceptions import (
+    DurableOperationError,
     OrphanedChildException,
     SuspendExecution,
     TimedSuspendExecution,
@@ -418,11 +419,26 @@ def _create_result(self) -> BatchResult[ResultType]:
                         )
                     )
                 case BranchStatus.FAILED:
+                    err = executable.error
+                    # Record the raw escaping type so first-run matches the
+                    # branch's FAIL checkpoint that replay reads back;
+                    # from_exception on the ChildContextError wrapper would
+                    # instead record the wrapper class name.
+                    error_object = (
+                        ErrorObject(
+                            message=err.message,
+                            type=err.error_type,
+                            data=err.data,
+                            stack_trace=err.stack_trace,
+                        )
+                        if isinstance(err, DurableOperationError)
+                        else ErrorObject.from_exception(err)
+                    )
                     batch_items.append(
                         BatchItem(
                             executable.index,
                             BatchItemStatus.FAILED,
-                            error=ErrorObject.from_exception(executable.error),
+                            error=error_object,
                         )
                     )
                 case (

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

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING, Generic, TypeVar
 
 from aws_durable_execution_sdk_python.exceptions import (
+    ChildContextError,
     InvalidStateError,
     SuspendExecution,
 )
@@ -260,8 +261,12 @@ def throw_if_error(self) -> None:
             (item.error for item in self.all if item.status is BatchItemStatus.FAILED),
             None,
         )
-        if first_error:
-            raise first_error.to_callable_runtime_error()
+        if first_error is not None:
+            # Each item runs in a child context, so a failed item surfaces as a
+            # ChildContextError wrapping the escaping error (reconstructed as its
+            # typed form where possible, with __cause__ set). Remaining errors
+            # stay in get_errors().
+            first_error.raise_as_operation_error(ChildContextError)
 
     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
 
@@ -266,25 +265,88 @@ 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 concrete class identifies the
+    operation kind (so callers can ``except StepError``); the escaping error is
+    preserved as ``__cause__`` on the first run and reconstructed on replay.
 
-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: Type name of the error that escaped the operation (the
+            original/inner error, e.g. ``"ValueError"``), not the operation
+            class. Defaults to the class name only when no escaping error is
+            known.
+        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
+
+    @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."""
+
+
+# Reconstruction registry for replay: only the SDK's own operation-error types.
+# Any other discriminator falls back to DurableOperationError in
+# from_error_fields, so we never call a constructor the SDK doesn't control.
+_DURABLE_OPERATION_ERROR_REGISTRY: dict[str, type[DurableOperationError]] = {
+    "StepError": StepError,
+    "InvokeError": InvokeError,
+    "ChildContextError": ChildContextError,
+    "WaitForConditionError": WaitForConditionError,
+}
 
 
 class StepInterruptedError(InvocationError):
@@ -397,37 +459,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."""