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 = 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.
+    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

@@ -7,6 +7,7 @@
 import threading
 from typing import Any, TYPE_CHECKING, cast
 
+from aws_durable_execution_sdk_python.lambda_service import ErrorObject
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
     InvocationEndInfo,
@@ -563,22 +564,28 @@ 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: datetime.datetime | None = None,
+        *,
+        error: ErrorObject | 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.
+            error: Diagnostic details recorded at the normalized span end.
         """
         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)
+            record_error_object_exception(span, error, timestamp=end_time)
             span.end(end_time=end_time)
             logger.debug("Ended OTel span: %s", span)
 
@@ -965,11 +972,16 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
 
         if info.error:
             span.set_status(StatusCode.ERROR, info.error.message or "")
-            record_error_object_exception(span, info.error)
+        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, info.end_time, error=info.error)
 
     def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
         """Called when a context or step operation starts user code.
@@ -1088,14 +1100,19 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
                 span.set_status(
                     StatusCode.ERROR, info.error.message if info.error else ""
                 )
-                record_error_object_exception(span, info.error)
             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)
+            self._end_span(
+                span_key,
+                end_timestamp,
+                error=info.error
+                if info.outcome is UserFunctionOutcome.FAILED
+                else None,
+            )
         # Restore the enclosing context by releasing the scope this user
         # function attached. Code that runs after this operation (e.g. between
         # steps in a child context) correlates to its enclosing operation

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