fix: OTel operation and attempt events lose the original exception.type

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py

@@ -0,0 +1,108 @@
+"""Reconstruct exceptions from serialized error metadata for span events.
+
+The SDK checkpoints an ``ErrorObject`` (message/type/data/stack_trace) for a
+failed operation or attempt. OpenTelemetry derives ``exception.type`` from the
+runtime class passed to ``Span.record_exception``, so constructing a generic
+``Exception`` collapses every failure to ``Exception`` and loses the original
+type. These helpers rebuild an exception that carries the recorded type and
+keep the original message and checkpointed stack trace.
+"""
+
+from __future__ import annotations
+
+import builtins
+import importlib
+from typing import Any
+
+from aws_durable_execution_sdk_python.lambda_service import ErrorObject
+
+
+_DEFAULT_MESSAGE = "Unknown error"
+
+
+def _resolve_exception_class(type_name: str) -> type[BaseException] | None:
+    """Resolve a wire ErrorType to a real exception class, if available."""
+    candidate: Any = getattr(builtins, type_name, None)
+    if isinstance(candidate, type) and issubclass(candidate, BaseException):
+        return candidate
+    if "." in type_name:
+        module_name, _, qualname = type_name.rpartition(".")
+        try:
+            module = importlib.import_module(module_name)
+        except ImportError:
+            return None
+        candidate = module
+        for part in qualname.split("."):
+            candidate = getattr(candidate, part, None)
+            if candidate is None:
+                return None
+        if isinstance(candidate, type) and issubclass(candidate, BaseException):
+            return candidate
+    return None
+
+
+def _synthesize_exception_class(type_name: str) -> type[BaseException]:
+    """Build an exception class whose qualified name matches ``type_name``.
+
+    Used when the original type is not importable. ``record_exception`` reads
+    ``__module__``/``__qualname__`` from the class, so a class carrying the
+    original name reproduces ``exception.type`` without rerunning user code.
+    """
+    if "." in type_name:
+        module_name, _, qualname = type_name.rpartition(".")
+    else:
+        module_name, qualname = "builtins", type_name
+    return type(qualname, (Exception,), {"__module__": module_name})
+
+
+def exception_from_error_object(error: ErrorObject) -> BaseException:
+    """Rebuild an exception that preserves the SDK-recorded error type.
+
+    Prefers a resolvable real class, then a synthesized class named after the
+    wire type, falling back to ``Exception`` only when no type is available.
+    """
+    message: str = error.message if error.message is not None else ""
+    type_name: str | None = error.type
+    if not type_name:
+        return Exception(message or _DEFAULT_MESSAGE)
+    resolved: type[BaseException] | None = _resolve_exception_class(type_name)
+    cls: type[BaseException] = resolved or _synthesize_exception_class(type_name)
+    try:
+        return cls(message)
+    except Exception:  # noqa: BLE001 - a class may reject positional args
+        exc: BaseException = cls.__new__(cls)
+        BaseException.__init__(exc, message)
+        return exc
+
+
+def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]:
+    """Return explicit event attributes for a checkpointed error.
+
+    OTel derives ``exception.message`` from ``str(exception)`` and auto-fills
+    ``exception.stacktrace`` from a live traceback; a rebuilt exception has
+    neither the exact message formatting (e.g. ``KeyError`` quotes its
+    argument) nor a traceback, so the recorded values are supplied explicitly.
+    """
+    attributes: dict[str, Any] = {}
+    if error.type is not None:
+        attributes["exception.type"] = error.type
+    if error.message is not None:
+        attributes["exception.message"] = error.message
+    if error.stack_trace:
+        attributes["exception.stacktrace"] = "\n".join(error.stack_trace)
+    return attributes
+
+
+def record_error_object_exception(span: Any, error: ErrorObject | None) -> None:
+    """Record a span exception that preserves the SDK's original error type.
+
+    ``error`` may be ``None`` for a failed outcome that carried no metadata; a
+    generic exception is recorded in that case.
+    """
+    if error is None:
+        span.record_exception(Exception(_DEFAULT_MESSAGE))
+        return
+    span.record_exception(
+        exception_from_error_object(error),
+        attributes=error_object_exception_attributes(error),
+    )

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py

@@ -87,6 +87,9 @@
     ensure_end_after_start,
 )
 from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot
+from aws_durable_execution_sdk_python_otel.exception_events import (
+    record_error_object_exception,
+)
 from aws_durable_execution_sdk_python_otel.execution_trace_context import (
     ExecutionTraceContext,
     canonical_trace_id,
@@ -891,9 +894,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
 
         if info.error:
             span.set_status(StatusCode.ERROR, info.error.message or "")
-            span.record_exception(
-                Exception(info.error.message or info.error.type or "Unknown error")
-            )
+            record_error_object_exception(span, info.error)
         elif info.status is OperationStatus.SUCCEEDED:
             span.set_status(StatusCode.OK)
 
@@ -1027,13 +1028,7 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
                 span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-                span.record_exception(
-                    Exception(
-                        (info.error.message or info.error.type)
-                        if info.error
-                        else "Unknown error"
-                    )
-                )
+                record_error_object_exception(span, info.error)
             else:
                 span.set_status(StatusCode.OK)
 

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py

@@ -55,6 +55,9 @@
 from aws_durable_execution_sdk_python_otel.durable_parent_span import (
     DurableParentSpan,
 )
+from aws_durable_execution_sdk_python_otel.exception_events import (
+    record_error_object_exception,
+)
 from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot
 from aws_durable_execution_sdk_python_otel.execution_trace_context import (
     ExecutionTraceContext,
@@ -962,9 +965,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
 
         if info.error:
             span.set_status(StatusCode.ERROR, info.error.message or "")
-            span.record_exception(
-                Exception(info.error.message or info.error.type or "Unknown error")
-            )
+            record_error_object_exception(span, info.error)
         elif info.status is OperationStatus.SUCCEEDED:
             span.set_status(StatusCode.OK)
 
@@ -1087,13 +1088,7 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
                 span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-                span.record_exception(
-                    Exception(
-                        (info.error.message or info.error.type)
-                        if info.error
-                        else "Unknown error"
-                    )
-                )
+                record_error_object_exception(span, info.error)
             else:
                 span.set_status(StatusCode.OK)
 

packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py

@@ -17,6 +17,9 @@
     OperationEndInfo,
     OperationStartInfo,
     OperationType,
+    UserFunctionEndInfo,
+    UserFunctionOutcome,
+    UserFunctionStartInfo,
 )
 from opentelemetry import context
 from opentelemetry.sdk.trace import TracerProvider
@@ -170,3 +173,160 @@ def test_retry_then_resume_exports_retrying(
     assert spans[0].context is not None
     assert spans[1].context is not None
     assert spans[0].context.trace_id == spans[1].context.trace_id
+
+
+def _start_invocation(plugin: Any) -> None:
+    plugin.on_invocation_start(
+        InvocationStartInfo(
+            request_id="request-1",
+            execution_arn=ARN,
+            execution_start_time=START,
+            is_first_invocation=True,
+        )
+    )
+
+
+def _end_invocation(plugin: Any) -> None:
+    plugin.on_invocation_end(
+        InvocationEndInfo(
+            request_id="request-1",
+            execution_arn=ARN,
+            execution_start_time=START,
+            is_first_invocation=True,
+            status=InvocationStatus.SUCCEEDED,
+        )
+    )
+
+
+def _operation_end_info(
+    error: ErrorObject | None, *, is_replayed: bool = False
+) -> OperationEndInfo:
+    return OperationEndInfo(
+        operation_id="wait",
+        name="wait",
+        parent_id=None,
+        operation_type=OperationType.WAIT,
+        sub_type=OperationSubType.WAIT,
+        start_time=START,
+        is_replayed=is_replayed,
+        end_time=END,
+        status=OperationStatus.FAILED,
+        error=error,
+    )
+
+
+def _run_failed_step_attempt(
+    plugin: Any, error: ErrorObject | None, *, is_replayed: bool = False
+) -> None:
+    operation: dict[str, Any] = dict(
+        operation_id="step-1",
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name="step-1",
+        parent_id=None,
+        start_time=START,
+        is_replayed=is_replayed,
+        status=OperationStatus.STARTED,
+        is_replay_children=False,
+        attempt=1,
+    )
+    plugin.on_user_function_start(UserFunctionStartInfo(**operation))
+    plugin.on_user_function_end(
+        UserFunctionEndInfo(
+            **operation,
+            outcome=UserFunctionOutcome.FAILED,
+            end_time=END,
+            error=error,
+        )
+    )
+
+
+def _exception_attributes(span: Any) -> Any:
+    assert len(span.events) == 1
+    assert span.events[0].name == "exception"
+    assert span.events[0].attributes is not None
+    return span.events[0].attributes
+
+
+CASES = [
+    ("ValueError", "example failure"),
+    ("KeyError", "missing key"),
+    ("TimeoutError", "deadline exceeded"),
+    ("custompkg.NotARealError", "boom"),
+]
+
+
+@pytest.mark.parametrize(("error_type", "message"), CASES)
+def test_operation_exception_event_preserves_original_type(
+    instrumented_view: InstrumentedView, error_type: str, message: str
+) -> None:
+    """The operation exception event must keep the SDK-supplied error type."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    plugin.on_operation_end(
+        _operation_end_info(ErrorObject(message, error_type, None, None))
+    )
+    _end_invocation(plugin)
+    span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
+    attributes = _exception_attributes(span)
+    assert attributes["exception.type"] == error_type
+    assert attributes["exception.message"] == message
+
+
+@pytest.mark.parametrize(("error_type", "message"), CASES)
+def test_attempt_exception_event_preserves_original_type(
+    instrumented_view: InstrumentedView, error_type: str, message: str
+) -> None:
+    """A failed STEP attempt event must keep the SDK-supplied error type."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    _run_failed_step_attempt(plugin, ErrorObject(message, error_type, None, None))
+    _end_invocation(plugin)
+    span = next(
+        s for s in exporter.get_finished_spans() if s.name == "step-1 attempt 1"
+    )
+    attributes = _exception_attributes(span)
+    assert attributes["exception.type"] == error_type
+    assert attributes["exception.message"] == message
+
+
+def test_replay_attempt_event_uses_stored_error_metadata(
+    instrumented_view: InstrumentedView,
+) -> None:
+    """A replayed attempt must recover type/message/stack from checkpoint data."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    stack_trace = [
+        "Traceback (most recent call last):",
+        '  File "handler.py", line 1, in handler',
+        "ValueError: example failure",
+    ]
+    _run_failed_step_attempt(
+        plugin,
+        ErrorObject("example failure", "ValueError", None, stack_trace),
+        is_replayed=True,
+    )
+    _end_invocation(plugin)
+    span = next(
+        s for s in exporter.get_finished_spans() if s.name == "step-1 attempt 1"
+    )
+    attributes = _exception_attributes(span)
+    assert attributes["exception.type"] == "ValueError"
+    assert attributes["exception.message"] == "example failure"
+    assert attributes["exception.stacktrace"] == "\n".join(stack_trace)
+
+
+def test_failed_operation_without_type_falls_back(
+    instrumented_view: InstrumentedView,
+) -> None:
+    """Without type metadata the event still carries the original message."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    plugin.on_operation_end(
+        _operation_end_info(ErrorObject("no type here", None, None, None))
+    )
+    _end_invocation(plugin)
+    span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
+    attributes = _exception_attributes(span)
+    assert attributes["exception.message"] == "no type here"
+    assert attributes["exception.type"] == "Exception"