From 26536b6d843dfea4e63502b2fdb2653e899d7c0f Mon Sep 17 00:00:00 2001
From: Frank Chen <65260095+zhongkechen@users.noreply.github.com>
Date: Tue, 27 Oct 2026 15:33:23 -0700
Subject: [PATCH 1/4] test: cover terminal OTel status and event times

---
 .../e2e/test_terminal_failure_bounds_int.py   | 133 ++++++++++++++++++
 .../tests/test_status_mapping.py              |  81 ++++++++++-
 2 files changed, 207 insertions(+), 7 deletions(-)
 create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py

diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py
new file mode 100644
index 000000000..c2d14256b
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py
@@ -0,0 +1,133 @@
+"""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="{}", 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"]
+        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()
diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
index 0b4bcc324..57cdc3c9d 100644
--- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
+++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
@@ -1,6 +1,7 @@
 """Shared status regressions through both public plugin lifecycles."""
 
 from collections.abc import Iterator
+from dataclasses import replace
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -69,13 +70,7 @@ def instrumented_view(request: pytest.FixtureRequest) -> Iterator[InstrumentedVi
 
 @pytest.mark.parametrize(
     "status",
-    [
-        OperationStatus.SUCCEEDED,
-        OperationStatus.FAILED,
-        OperationStatus.CANCELLED,
-        OperationStatus.TIMED_OUT,
-        OperationStatus.STOPPED,
-    ],
+    list(OperationStatus),
 )
 @pytest.mark.parametrize("with_error", [False, True])
 @pytest.mark.parametrize("started_here", [False, True])
@@ -129,6 +124,8 @@ def test_operation_status_and_exception(
     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
         )
@@ -144,6 +141,76 @@ def test_operation_status_and_exception(
         assert not span.events
 
 
+@pytest.mark.parametrize("start_time", [None, START, END])
+@pytest.mark.parametrize("end_time", [END, None])
+def test_terminal_exception_uses_normalized_span_end(
+    instrumented_view: InstrumentedView,
+    start_time: datetime | None,
+    end_time: datetime | None,
+) -> None:
+    """Delayed and checkpointless completions retain bounded diagnostic times."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    plugin.on_operation_end(
+        replace(
+            _operation_end_info(ErrorObject("stored failure", "ValueError", None, None)),
+            operation_type=OperationType.CONTEXT,
+            sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT,
+            start_time=start_time,
+            end_time=end_time,
+        )
+    )
+    _end_invocation(plugin)
+    span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
+    (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
+
+
+@pytest.mark.parametrize("with_error", [False, True])
+@pytest.mark.parametrize("end_time", [START, END, None])
+def test_failed_attempt_exception_uses_normalized_span_end(
+    instrumented_view: InstrumentedView,
+    with_error: bool,
+    end_time: datetime | None,
+) -> None:
+    """An attempt failure reports its outcome without fabricating diagnostics."""
+    plugin, exporter = instrumented_view
+    _start_invocation(plugin)
+    operation: dict[str, Any] = dict(
+        operation_id="attempt",
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name="attempt",
+        parent_id=None,
+        start_time=START,
+        is_replayed=False,
+        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_time,
+            error=ErrorObject("stored failure", "ValueError", None, None)
+            if with_error
+            else None,
+        )
+    )
+    _end_invocation(plugin)
+    span = next(s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1")
+    assert span.status.status_code is StatusCode.ERROR
+    if with_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
+    else:
+        assert not span.events
+
+
 def test_retry_then_resume_exports_retrying(
     instrumented_view: InstrumentedView,
 ) -> None:

From 82fc8767225cbbf4504a7c89ea033afc2b306b37 Mon Sep 17 00:00:00 2001
From: Frank Chen <65260095+zhongkechen@users.noreply.github.com>
Date: Tue, 27 Oct 2026 15:36:08 -0700
Subject: [PATCH 2/4] fix: align terminal OTel status and event times

---
 .../README.md                                 | 27 +++++++++++++----
 .../exception_events.py                       | 11 ++++---
 .../execution_plugin.py                       | 21 ++++++++++----
 .../invocation_plugin.py                      | 29 +++++++++++++++----
 .../e2e/test_terminal_failure_bounds_int.py   |  5 ++--
 .../tests/test_status_mapping.py              |  8 +++--
 6 files changed, 77 insertions(+), 24 deletions(-)

diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md
index 8e9ed0816..d9c503187 100644
--- a/packages/aws-durable-execution-sdk-python-otel/README.md
+++ b/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
 
diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py
index 872a7bedf..67156b63b 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py
+++ b/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,
     )
diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py
index 356990ede..2c52e294b 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py
+++ b/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
diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py
index 4eab6c7c3..81a2e00d3 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py
+++ b/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,8 +564,12 @@ 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:
@@ -572,6 +577,7 @@ def _end_span(
                 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:
@@ -579,6 +585,7 @@ def _end_span(
         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
diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py
index c2d14256b..79dd2eda4 100644
--- a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py
+++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py
@@ -95,7 +95,7 @@ def body(_event: Any, context: DurableContext) -> str:
             poll_interval=0.01,
         ) as runner:
             runner.register_function("failing-target", failing_function)
-            result = runner.run(input="{}", timeout=15)
+            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"]
@@ -125,7 +125,8 @@ def body(_event: Any, context: DurableContext) -> str:
                 "actual invoke failure",
             }
         invocations = [span for span in spans if span.name == "Invocation"]
-        assert len(invocations) == (2 if resume else 1)
+        # 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
diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
index 57cdc3c9d..eb37d24f8 100644
--- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
+++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
@@ -153,7 +153,9 @@ def test_terminal_exception_uses_normalized_span_end(
     _start_invocation(plugin)
     plugin.on_operation_end(
         replace(
-            _operation_end_info(ErrorObject("stored failure", "ValueError", None, None)),
+            _operation_end_info(
+                ErrorObject("stored failure", "ValueError", None, None)
+            ),
             operation_type=OperationType.CONTEXT,
             sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT,
             start_time=start_time,
@@ -201,7 +203,9 @@ def test_failed_attempt_exception_uses_normalized_span_end(
         )
     )
     _end_invocation(plugin)
-    span = next(s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1")
+    span = next(
+        s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1"
+    )
     assert span.status.status_code is StatusCode.ERROR
     if with_error:
         (event,) = span.events

From 28338a67b25b1d90c616186e75bf4044c27e5724 Mon Sep 17 00:00:00 2001
From: Frank Chen <65260095+zhongkechen@users.noreply.github.com>
Date: Thu, 29 Oct 2026 13:05:03 -0700
Subject: [PATCH 3/4] refactor: simplify terminal span completion

Normalize terminal timestamps in lifecycle handlers so exception events
and span ends use the same value. Require an explicit exception timestamp
and remove error plumbing and the duplicate equal-time adjustment.

Keep ordered fallback timestamps for invocation cleanup. Cover cleanup
when an incomplete attempt starts ahead of the local clock, and verify
that the new regression rejects removal of the fallback.
---
 .../exception_events.py                       |  2 +-
 .../execution_plugin.py                       |  4 +--
 .../invocation_plugin.py                      | 35 ++++++-------------
 .../tests/test_invocation_plugin.py           | 34 +++++++++++++++++-
 4 files changed, 47 insertions(+), 28 deletions(-)

diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py
index 67156b63b..aab273056 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py
+++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py
@@ -41,7 +41,7 @@ def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]:
 
 
 def record_error_object_exception(
-    span: Any, error: ErrorObject | None, *, timestamp: int | None = None
+    span: Any, error: ErrorObject | None, *, timestamp: int
 ) -> None:
     """Record a span exception that preserves the SDK's original error type.
 
diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py
index 2c52e294b..265c01074 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py
+++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py
@@ -892,7 +892,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
             span_id_override=span_id_override,
         )
 
-        end_timestamp = _to_otel_timestamp(
+        end_timestamp = to_otel_timestamp(
             end_time or datetime.datetime.now(datetime.UTC)
         )
         if info.error:
@@ -1033,7 +1033,7 @@ 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_timestamp = to_otel_timestamp(
                 end_time or datetime.datetime.now(datetime.UTC)
             )
             if info.outcome is UserFunctionOutcome.FAILED:
diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py
index 81a2e00d3..d54b97980 100644
--- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py
+++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py
@@ -7,7 +7,6 @@
 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,
@@ -564,29 +563,22 @@ def _start_span(
         return span
 
     def _end_span(
-        self,
-        operation_id: str | None,
-        end_timestamp: datetime.datetime | None = None,
-        *,
-        error: ErrorObject | 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.
-            error: Diagnostic details recorded at the normalized span end.
+            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)
-            record_error_object_exception(span, error, timestamp=end_time)
-            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)
 
     # ------------------------------------------------------------------
@@ -970,8 +962,10 @@ 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, timestamp=end_timestamp)
         elif info.status in {
             OperationStatus.FAILED,
             OperationStatus.TIMED_OUT,
@@ -981,7 +975,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None:
         elif info.status is OperationStatus.SUCCEEDED:
             span.set_status(StatusCode.OK)
 
-        self._end_span(info.operation_id, info.end_time, error=info.error)
+        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.
@@ -1096,23 +1090,16 @@ 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, 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,
-                error=info.error
-                if info.outcome is UserFunctionOutcome.FAILED
-                else None,
-            )
+            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
         # steps in a child context) correlates to its enclosing operation
diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py
index 27547c76c..09a132dc7 100644
--- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py
+++ b/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()

From ef406b5783180645974afbd41cf3e1bf40ffd6b6 Mon Sep 17 00:00:00 2001
From: Frank Chen <65260095+zhongkechen@users.noreply.github.com>
Date: Thu, 29 Oct 2026 13:16:29 -0700
Subject: [PATCH 4/4] test: verify durable end timestamp selection

Check exact normalized durable ends for operations and attempts, including
equal, reversed, and missing start times. Bound fallback ends by the
callback clock only when the durable end is absent.

Keep exception/end agreement assertions and verify that forcing wall-clock
time or removing the clock fallback is rejected in each handler.

Addresses review comment r4417357787.
---
 .../tests/test_status_mapping.py              | 52 +++++++++++++++++--
 1 file changed, 48 insertions(+), 4 deletions(-)

diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
index eb37d24f8..cbf47a88a 100644
--- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
+++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py
@@ -141,16 +141,26 @@ def test_operation_status_and_exception(
         assert not span.events
 
 
-@pytest.mark.parametrize("start_time", [None, START, END])
+@pytest.mark.parametrize(
+    ("start_time", "expected_durable_end"),
+    [
+        (None, END + timedelta(microseconds=1)),
+        (START, END),
+        (END, END + timedelta(microseconds=1)),
+        (END + timedelta(seconds=1), END + timedelta(seconds=1, microseconds=1)),
+    ],
+)
 @pytest.mark.parametrize("end_time", [END, None])
 def test_terminal_exception_uses_normalized_span_end(
     instrumented_view: InstrumentedView,
     start_time: datetime | None,
+    expected_durable_end: datetime,
     end_time: datetime | None,
 ) -> None:
-    """Delayed and checkpointless completions retain bounded diagnostic times."""
+    """Execution view preserves durable ends, including delayed completions."""
     plugin, exporter = instrumented_view
     _start_invocation(plugin)
+    before_callback = datetime.now(UTC)
     plugin.on_operation_end(
         replace(
             _operation_end_info(
@@ -162,19 +172,40 @@ def test_terminal_exception_uses_normalized_span_end(
             end_time=end_time,
         )
     )
+    after_callback = datetime.now(UTC)
     _end_invocation(plugin)
     span = next(s for s in exporter.get_finished_spans() if s.name == "wait")
     (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
+    if isinstance(plugin, ExecutionOtelPlugin):
+        if end_time is None:
+            assert (
+                int(before_callback.timestamp() * 1_000_000_000)
+                <= span.end_time
+                <= int(after_callback.timestamp() * 1_000_000_000)
+            )
+        else:
+            assert span.end_time == int(
+                expected_durable_end.timestamp() * 1_000_000_000
+            )
 
 
 @pytest.mark.parametrize("with_error", [False, True])
-@pytest.mark.parametrize("end_time", [START, END, None])
+@pytest.mark.parametrize(
+    ("end_time", "expected_durable_end"),
+    [
+        (START - timedelta(seconds=1), START + timedelta(microseconds=1)),
+        (START, START + timedelta(microseconds=1)),
+        (END, END),
+        (None, None),
+    ],
+)
 def test_failed_attempt_exception_uses_normalized_span_end(
     instrumented_view: InstrumentedView,
     with_error: bool,
     end_time: datetime | None,
+    expected_durable_end: datetime | None,
 ) -> None:
     """An attempt failure reports its outcome without fabricating diagnostics."""
     plugin, exporter = instrumented_view
@@ -192,6 +223,7 @@ def test_failed_attempt_exception_uses_normalized_span_end(
         attempt=1,
     )
     plugin.on_user_function_start(UserFunctionStartInfo(**operation))
+    before_callback = datetime.now(UTC)
     plugin.on_user_function_end(
         UserFunctionEndInfo(
             **operation,
@@ -202,14 +234,26 @@ def test_failed_attempt_exception_uses_normalized_span_end(
             else None,
         )
     )
+    after_callback = datetime.now(UTC)
     _end_invocation(plugin)
     span = next(
         s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1"
     )
     assert span.status.status_code is StatusCode.ERROR
+    assert span.start_time is not None and span.end_time is not None
+    if isinstance(plugin, ExecutionOtelPlugin):
+        if expected_durable_end is None:
+            assert (
+                int(before_callback.timestamp() * 1_000_000_000)
+                <= span.end_time
+                <= int(after_callback.timestamp() * 1_000_000_000)
+            )
+        else:
+            assert span.end_time == int(
+                expected_durable_end.timestamp() * 1_000_000_000
+            )
     if with_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
     else:
         assert not span.events