refactor: share operation span status mapping

packages/aws-durable-execution-sdk-python-otel/README.md

@@ -416,13 +416,18 @@ Failure status does not fabricate an exception when diagnostics are absent,
 including for failed step attempts. A caught operation failure leaves the
 enclosing Invocation and Workflow spans successful if the handler succeeds.
 
-Terminal operation and attempt exception events use the same normalized end
-timestamp as their span, including delayed or replayed completion delivery.
-Execution view retains durable timing and normalizes an end at or before its
-start; Invocation view retains its invocation-local timestamp ordering. A
-checkpointless context with no durable start uses its completion time as the
-lower bound in Execution view. When an end timestamp is absent, the current time
-is chosen once for both the event and span end.
+Operation and step-attempt exception events have the same timestamp as their
+span end, including when completions arrive late or are observed on replay.
+
+Execution view uses the end time supplied by the SDK, extending operation spans
+to include later observed child ends. An end at or before the start moves to
+1 µs after the start.
+
+Invocation view keeps timestamps in order within an invocation. An end at or
+before the last timestamp it used moves to 1 µs after that timestamp.
+
+When no end time is available, the view reads the clock once for the event and
+span end. Invocation view still applies its timestamp ordering.
 
 ### Invocation context isolation
 

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

@@ -43,7 +43,6 @@
     InvocationStartInfo,
     OperationEndInfo,
     OperationStartInfo,
-    OperationStatus,
     OperationType,
     UserFunctionEndInfo,
     UserFunctionOutcome,
@@ -95,6 +94,9 @@
     canonical_trace_id,
 )
 from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig
+from aws_durable_execution_sdk_python_otel.operation_span_status import (
+    set_operation_span_status,
+)
 from aws_durable_execution_sdk_python_otel.log_filter import (
     bind_invocation,
     install_log_filter,
@@ -895,17 +897,8 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
         end_timestamp = to_otel_timestamp(
             end_time or datetime.datetime.now(datetime.UTC)
         )
-        if info.error:
-            span.set_status(StatusCode.ERROR, info.error.message or "")
-            record_error_object_exception(span, info.error, timestamp=end_timestamp)
-        elif info.status in {
-            OperationStatus.FAILED,
-            OperationStatus.TIMED_OUT,
-            OperationStatus.STOPPED,
-        }:
-            span.set_status(StatusCode.ERROR)
-        elif info.status is OperationStatus.SUCCEEDED:
-            span.set_status(StatusCode.OK)
+        set_operation_span_status(span, info)
+        record_error_object_exception(span, info.error, timestamp=end_timestamp)
 
         self._note_parent_end(info.parent_id, end_time)
         popped = self._pop_span(info.operation_id)

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

@@ -14,7 +14,6 @@
     InvocationStartInfo,
     OperationEndInfo,
     OperationStartInfo,
-    OperationStatus,
     OperationType,
     UserFunctionEndInfo,
     UserFunctionOutcome,
@@ -69,6 +68,9 @@
     unbind_invocation,
 )
 from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig
+from aws_durable_execution_sdk_python_otel.operation_span_status import (
+    set_operation_span_status,
+)
 from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider
 from aws_durable_execution_sdk_python_otel.propagation import propagation_metadata
 from aws_durable_execution_sdk_python_otel.timestamps import to_otel_timestamp
@@ -963,17 +965,8 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
             span.set_attributes(self._extract_attributes(info))
 
         end_timestamp = self._next_ordered_timestamp(info.end_time)
-        if info.error:
-            span.set_status(StatusCode.ERROR, info.error.message or "")
-            record_error_object_exception(span, info.error, timestamp=end_timestamp)
-        elif info.status in {
-            OperationStatus.FAILED,
-            OperationStatus.TIMED_OUT,
-            OperationStatus.STOPPED,
-        }:
-            span.set_status(StatusCode.ERROR)
-        elif info.status is OperationStatus.SUCCEEDED:
-            span.set_status(StatusCode.OK)
+        set_operation_span_status(span, info)
+        record_error_object_exception(span, info.error, timestamp=end_timestamp)
 
         self._end_span(info.operation_id, end_timestamp)
 

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

@@ -0,0 +1,20 @@
+"""OpenTelemetry operation status mapping shared by both views."""
+
+from aws_durable_execution_sdk_python.lambda_service import OperationStatus
+from aws_durable_execution_sdk_python.plugin import OperationEndInfo
+from opentelemetry.trace import Span, StatusCode
+
+
+_FAILURE_STATUSES = frozenset(
+    {OperationStatus.FAILED, OperationStatus.TIMED_OUT, OperationStatus.STOPPED}
+)
+
+
+def set_operation_span_status(span: Span, info: OperationEndInfo) -> None:
+    """Apply the durable outcome, with supplied error details taking precedence."""
+    if info.error:
+        span.set_status(StatusCode.ERROR, info.error.message or "")
+    elif info.status in _FAILURE_STATUSES:
+        span.set_status(StatusCode.ERROR)
+    elif info.status is OperationStatus.SUCCEEDED:
+        span.set_status(StatusCode.OK)

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

@@ -43,6 +43,16 @@
 ARN = (
     "arn:aws:lambda:us-west-2:123456789012:function:status:1/durable-execution/test/id"
 )
+_STATUS_WITHOUT_ERROR: dict[OperationStatus, StatusCode] = {
+    OperationStatus.STARTED: StatusCode.UNSET,
+    OperationStatus.PENDING: StatusCode.UNSET,
+    OperationStatus.READY: StatusCode.UNSET,
+    OperationStatus.SUCCEEDED: StatusCode.OK,
+    OperationStatus.FAILED: StatusCode.ERROR,
+    OperationStatus.CANCELLED: StatusCode.UNSET,
+    OperationStatus.TIMED_OUT: StatusCode.ERROR,
+    OperationStatus.STOPPED: StatusCode.ERROR,
+}
 
 
 InstrumentedView = tuple[
@@ -121,15 +131,7 @@ def test_operation_status_and_exception(
     span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
     assert span.attributes is not None
     assert span.attributes["durable.operation.status"] == status.value
-    expected = (
-        StatusCode.ERROR
-        if with_error
-        or status
-        in {OperationStatus.FAILED, OperationStatus.TIMED_OUT, OperationStatus.STOPPED}
-        else (
-            StatusCode.OK if status is OperationStatus.SUCCEEDED else StatusCode.UNSET
-        )
-    )
+    expected = StatusCode.ERROR if with_error else _STATUS_WITHOUT_ERROR[status]
     assert span.status.status_code is expected
     if with_error:
         assert span.status.description == "deadline exceeded"
@@ -495,6 +497,8 @@ def test_failed_operation_without_type_or_message_uses_fallback(
     plugin.on_operation_end(_operation_end_info(ErrorObject(None, None, None, None)))
     _end_invocation(plugin)
     span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.status.description == ""
     attributes = _exception_attributes(span)
     assert attributes["exception.message"] == "Unknown error"
     assert attributes["exception.type"] == "Exception"