fix: correct terminal telemetry status and timing

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

@@ -401,11 +401,28 @@ each durable span in the same invocation.
 ### Status attributes
 
 `durable.invocation.status` uses `RETRYING` when the core plugin hook reports
-`InvocationStatus.RETRY`. The core enum remains unchanged. Operation spans use
-OTel `OK` only for `SUCCEEDED`, `ERROR` when error details are delivered, and
-`UNSET` for other outcomes without error details, including `FAILED`,
-`CANCELLED`, `TIMED_OUT`, and `STOPPED`. The original durable operation status
-remains in `durable.operation.status`.
+`InvocationStatus.RETRY`, with OTel `UNSET`. The core enum remains unchanged.
+Operation spans use the following mapping in both views:
+
+| Durable operation outcome | Error details | OTel status | Exception event |
+| --- | --- | --- | --- |
+| `SUCCEEDED` | Absent | `OK` | None |
+| `FAILED`, `TIMED_OUT`, `STOPPED` | Absent | `ERROR` | None |
+| `CANCELLED` and other non-success outcomes | Absent | `UNSET` | None |
+| Any outcome | Present | `ERROR` | Available diagnostics |
+
+The original durable operation status remains in `durable.operation.status`.
+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.
 
 ### Invocation context isolation
 

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

@@ -40,16 +40,19 @@ def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]:
     return attributes
 
 
-def record_error_object_exception(span: Any, error: ErrorObject | None) -> None:
+def record_error_object_exception(
+    span: Any, error: ErrorObject | None, *, timestamp: int
+) -> 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.
+    A failed outcome without an error object has no diagnostic event. Terminal
+    callers supply the normalized span-end timestamp so delayed delivery does
+    not place an event outside the span.
     """
     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),
+        timestamp=timestamp,
     )

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

@@ -892,16 +892,25 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
             span_id_override=span_id_override,
         )
 
+        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)
+            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)
 
         self._note_parent_end(info.parent_id, end_time)
         popped = self._pop_span(info.operation_id)
         if popped is not None:
-            popped.end(end_time=_to_otel_timestamp(end_time))
+            popped.end(end_time=end_timestamp)
 
     def _start_span(
         self,
@@ -1024,18 +1033,20 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
             and info.outcome is not UserFunctionOutcome.INCOMPLETE
         ):
             span.set_attributes(self._operation_attributes(info))
+            end_timestamp = to_otel_timestamp(
+                end_time or datetime.datetime.now(datetime.UTC)
+            )
             if info.outcome is UserFunctionOutcome.FAILED:
                 span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-                record_error_object_exception(span, info.error)
+                record_error_object_exception(span, info.error, timestamp=end_timestamp)
             else:
                 span.set_status(StatusCode.OK)
 
-            end_time = ensure_end_after_start(info.start_time, info.end_time)
             popped = self._pop_span(key)
             if popped is not None:
-                popped.end(end_time=_to_otel_timestamp(end_time))
+                popped.end(end_time=end_timestamp)
 
         # Restore the enclosing context by releasing the scope this user
         # function attached, so the parent operation (or, at the top level, the

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

@@ -563,23 +563,22 @@ def _start_span(
         return span
 
     def _end_span(
-        self, operation_id: str | None, end_timestamp: datetime.datetime | None = None
-    ):
+        self, operation_id: str | None, end_timestamp: int | None = None
+    ) -> None:
         """End and unregister the active span for an operation ID.
 
         Args:
             operation_id: Durable operation ID, or ``None`` for the invocation
                 span.
-            end_timestamp: Optional durable end timestamp to use as the span end
-                time. When omitted, OpenTelemetry uses the current time.
+            end_timestamp: Normalized nanoseconds; omit for the next ordered time.
         """
         logger.debug("Ending OTel span: operation_id=%s", operation_id)
         with self._operation_spans_lock:
             span = self._operation_spans.pop(operation_id, None)
         if span:
-            # the span is not going to be populated if it has the same end_time and start_time
-            end_time = self._next_ordered_timestamp(end_timestamp)
-            span.end(end_time=end_time)
+            if end_timestamp is None:
+                end_timestamp = self._next_ordered_timestamp()
+            span.end(end_time=end_timestamp)
             logger.debug("Ended OTel span: %s", span)
 
     # ------------------------------------------------------------------
@@ -963,13 +962,20 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
         else:
             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)
+            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)
 
-        self._end_span(info.operation_id, info.end_time)
+        self._end_span(info.operation_id, end_timestamp)
 
     def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
         """Called when a context or step operation starts user code.
@@ -1084,17 +1090,15 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
             and info.outcome is not UserFunctionOutcome.INCOMPLETE
         ):
             span.set_attributes(self._extract_attributes(info))
+            end_timestamp = self._next_ordered_timestamp(info.end_time)
             if info.outcome is UserFunctionOutcome.FAILED:
                 span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-                record_error_object_exception(span, info.error)
+                record_error_object_exception(span, info.error, timestamp=end_timestamp)
             else:
                 span.set_status(StatusCode.OK)
 
-            end_timestamp = info.end_time
-            if end_timestamp is not None and end_timestamp == info.start_time:
-                end_timestamp += datetime.timedelta(microseconds=1)
             self._end_span(span_key, end_timestamp)
         # Restore the enclosing context by releasing the scope this user
         # function attached. Code that runs after this operation (e.g. between

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

@@ -0,0 +1,134 @@
+"""Caught operation failures preserve telemetry through real suspension/replay."""
+
+import json
+from typing import Any
+
+import pytest
+from aws_durable_execution_sdk_python import DurableContext, durable_execution
+from aws_durable_execution_sdk_python.config import ChildConfig, Duration
+from aws_durable_execution_sdk_python.exceptions import ChildContextError, InvokeError
+from aws_durable_execution_sdk_python.plugin import (
+    DurableInstrumentationPlugin,
+    InvocationStartInfo,
+    OperationEndInfo,
+)
+from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import StatusCode
+
+from aws_durable_execution_sdk_python_otel import (
+    ExecutionOtelPluginFactory,
+    InvocationOtelPluginFactory,
+    OtelPluginConfig,
+)
+
+
+@pytest.mark.parametrize(
+    "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory]
+)
+@pytest.mark.parametrize("failure_kind", ["child", "virtual", "invoke"])
+@pytest.mark.parametrize("resume", [False, True])
+def test_caught_failure_keeps_events_bounded_and_workflow_successful(
+    view: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory],
+    failure_kind: str,
+    resume: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False)
+    provider = TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    factory = view(
+        OtelPluginConfig(
+            tracer_provider=provider,
+            enrich_logger=False,
+            context_extractor=lambda _: None,
+        )
+    )
+    saved_calls: list[str] = []
+    failure_calls: list[str] = []
+    observations: list[OperationEndInfo] = []
+
+    class Observe(DurableInstrumentationPlugin):
+        def on_operation_end(self, info: OperationEndInfo) -> None:
+            if info.name == "caught-failure":
+                observations.append(info)
+
+    class ObserveFactory:
+        def create_plugin(self, _info: InvocationStartInfo) -> Observe:
+            return Observe()
+
+    def saved_step(_context: Any) -> str:
+        saved_calls.append("saved")
+        return "saved"
+
+    def failing_child(_context: DurableContext) -> str:
+        failure_calls.append("failed")
+        raise ValueError("actual child failure")
+
+    def failing_function(_event: Any, _context: Any) -> None:
+        failure_calls.append("failed")
+        raise ValueError("actual invoke failure")
+
+    def body(_event: Any, context: DurableContext) -> str:
+        saved = context.step(saved_step, name="saved-step")
+        try:
+            if failure_kind == "invoke":
+                context.invoke("failing-target", {}, name="caught-failure")
+            else:
+                context.run_in_child_context(
+                    failing_child,
+                    name="caught-failure",
+                    config=ChildConfig(is_virtual=failure_kind == "virtual"),
+                )
+        except (ChildContextError, InvokeError):
+            if resume:
+                context.wait(Duration.from_seconds(1), name="replay-wait")
+            return saved
+        raise AssertionError("Expected the original operation failure")
+
+    try:
+        with DurableFunctionTestRunner(
+            handler=durable_execution(body, plugins=[factory, ObserveFactory()]),
+            poll_interval=0.01,
+        ) as runner:
+            runner.register_function("failing-target", failing_function)
+            result = runner.run(input="{}", execution_timeout=15)
+        assert result.status.value == "SUCCEEDED"
+        assert result.result is not None and json.loads(result.result) == "saved"
+        assert saved_calls == ["saved"]
+        # Virtual contexts intentionally execute on replay; checkpointed child
+        # failures and invoke outcomes must return their stored failure instead.
+        assert len(failure_calls) == (2 if resume and failure_kind == "virtual" else 1)
+        assert observations
+        if failure_kind == "virtual":
+            assert all(info.start_time is None for info in observations)
+        spans = exporter.get_finished_spans()
+        failed = [
+            span
+            for span in spans
+            if span.name == "caught-failure"
+            and span.attributes is not None
+            and span.attributes.get("durable.operation.status") == "FAILED"
+        ]
+        assert failed
+        for span in failed:
+            assert span.status.status_code is StatusCode.ERROR
+            (event,) = span.events
+            assert span.start_time is not None and span.end_time is not None
+            assert span.start_time <= event.timestamp == span.end_time
+            assert event.attributes is not None
+            assert event.attributes["exception.message"] in {
+                "actual child failure",
+                "actual invoke failure",
+            }
+        invocations = [span for span in spans if span.name == "Invocation"]
+        # An invoke can add its own suspension before the explicit replay wait.
+        assert len(invocations) >= (2 if resume else 1)
+        assert all(span.status.status_code is StatusCode.OK for span in invocations)
+        (workflow,) = [span for span in spans if span.name == "Workflow"]
+        assert workflow.status.status_code is StatusCode.OK
+    finally:
+        provider.shutdown()

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

@@ -6,7 +6,7 @@
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 
 import opentelemetry.context as otel_context
@@ -2061,6 +2061,38 @@ def test_incomplete_attempt_is_marked_when_invocation_ends():
     assert attempt.attributes["durable.span.truncated_at_invocation_boundary"] is True
 
 
+def test_invocation_cleanup_keeps_future_attempt_bounded() -> None:
+    """Cleanup remains ordered when an attempt started ahead of the local clock."""
+    plugin, exporter = _create_plugin()
+    start_info = UserFunctionStartInfo(
+        operation_id="future-attempt",
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name="future-attempt",
+        parent_id=None,
+        start_time=datetime.now(UTC) + timedelta(days=1),
+        is_replayed=False,
+        status=OperationStatus.STARTED,
+        is_replay_children=False,
+        attempt=1,
+    )
+    plugin.on_invocation_start(_invocation_start_info())
+    plugin.on_user_function_start(start_info)
+    plugin.on_user_function_end(
+        UserFunctionEndInfo.from_start_info(
+            start_info, error=None, outcome=UserFunctionOutcome.INCOMPLETE
+        )
+    )
+    plugin.on_invocation_end(_invocation_end_info(status=InvocationStatus.PENDING))
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    attempt = spans["future-attempt attempt 1"]
+    invocation = spans["Invocation"]
+    assert attempt.start_time is not None and attempt.end_time is not None
+    assert invocation.end_time is not None
+    assert attempt.start_time < attempt.end_time < invocation.end_time
+
+
 def test_workflow_span_name_is_configurable():
     """The Workflow span name can be overridden via constructor kwarg."""
     exporter = InMemorySpanExporter()