fix: preserve empty callback error history

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

@@ -172,6 +172,16 @@ flow, pagination, and a map of the code.
 
 ### Error Handling
 
+A callback failure with omitted diagnostics or an empty `ErrorObject` has no
+error details in SDK-facing operation state, in both memory and filesystem stores.
+Detailed callback-failure history preserves `Error.Payload: {}` with
+`Truncated: false`; history requested without execution data omits the payload and
+sets `Truncated: true`. Reconstructing operations from detailed history restores
+the same absent diagnostics. Present diagnostic fields remain intact, including
+explicit empty strings and empty stack-trace lists. The callback still fails and
+raises the existing `CallbackExternalError`, including its default message when
+diagnostics are absent. Timeout history is unchanged.
+
 The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see:
 
 - [Error Response Documentation](docs/error-responses.md)

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py

@@ -709,7 +709,9 @@ def complete_callback_failure(
             updated_callback_details = None
             if operation.callback_details:
                 updated_callback_details = replace(
-                    operation.callback_details, error=error
+                    operation.callback_details,
+                    # Match SDK state after serialization: Error={} has no diagnostics.
+                    error=error if error.to_dict() else None,
                 )
 
             self.operations[index] = replace(

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py

@@ -675,7 +675,8 @@ class EventError:
     @classmethod
     def from_dict(cls, data: dict) -> EventError:
         payload = None
-        if payload_data := data.get("Payload"):
+        payload_data = data.get("Payload")
+        if payload_data is not None:
             payload = ErrorObject.from_dict(payload_data)
 
         return cls(
@@ -2249,6 +2250,14 @@ def create_callback_event_failed(cls, context: EventCreationContext) -> Event:
             if callback_details
             else None
         )
+        if (
+            context.include_execution_data
+            and callback_details is not None
+            and callback_details.error is None
+        ):
+            # Detailed failure history retains empty payload presence. Do not
+            # mutate the absent diagnostics in the SDK-facing operation state.
+            event_error = EventError(payload=ErrorObject.from_dict({}), truncated=False)
         return cls(
             event_type=EventType.CALLBACK_FAILED.value,
             event_timestamp=context.end_timestamp,
@@ -2606,6 +2615,11 @@ class HistoryEventTypeConfig:
 }
 
 
+def _operation_error(error: ErrorObject | None) -> ErrorObject | None:
+    """Normalize history's empty payload to absent SDK-state diagnostics."""
+    return error if error is not None and error.to_dict() else None
+
+
 def events_to_operations(events: list[Event]) -> list[Operation]:
     """Convert a list of history events into operations.
 
@@ -2750,7 +2764,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]:
                 callback_details=CallbackDetails(
                     callback_id=callback_id,
                     result=result,
-                    error=error,
+                    error=_operation_error(error),
                 ),
             )
 
@@ -2791,7 +2805,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]:
                 operation,
                 step_details=StepDetails(
                     result=result_val,
-                    error=error_val,
+                    error=_operation_error(error_val),
                     attempt=attempt,
                     next_attempt_timestamp=next_attempt_ts,
                 ),
@@ -2824,7 +2838,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]:
                     operation,
                     context_details=ContextDetails(
                         result=None,
-                        error=event.context_failed_details.error.payload,
+                        error=_operation_error(
+                            event.context_failed_details.error.payload
+                        ),
                     ),
                 )
 
@@ -2849,7 +2865,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]:
                     operation,
                     chained_invoke_details=ChainedInvokeDetails(
                         result=None,
-                        error=event.chained_invoke_failed_details.error.payload,
+                        error=_operation_error(
+                            event.chained_invoke_failed_details.error.payload
+                        ),
                     ),
                 )
 

packages/aws-durable-execution-sdk-python-testing/tests/e2e/empty_callback_error_test.py

@@ -0,0 +1,176 @@
+"""Real callback failure keeps runtime diagnostics and history consistent."""
+
+from datetime import datetime
+from pathlib import Path
+import time
+from typing import Any
+
+import pytest
+
+from aws_durable_execution_sdk_python import (
+    DurableContext,
+    durable_execution,
+    durable_step,
+)
+from aws_durable_execution_sdk_python.exceptions import (
+    CallbackError,
+    CallbackExternalError,
+)
+from aws_durable_execution_sdk_python.execution import InvocationStatus
+from aws_durable_execution_sdk_python.lambda_service import (
+    ErrorObject,
+    Operation,
+    OperationStatus,
+)
+from aws_durable_execution_sdk_python_testing.execution import Execution
+from aws_durable_execution_sdk_python_testing.model import (
+    GetDurableExecutionHistoryResponse,
+    events_to_operations,
+)
+from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner
+from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore
+from aws_durable_execution_sdk_python_testing.stores.filesystem import (
+    FileSystemExecutionStore,
+)
+from aws_durable_execution_sdk_python_testing.stores.memory import (
+    InMemoryExecutionStore,
+)
+
+
+@pytest.mark.parametrize("filesystem", [False, True])
+@pytest.mark.parametrize("error_kind", ["omitted", "empty", "rich"])
+def test_live_callback_failure_normalizes_before_persistence(
+    filesystem: bool,
+    error_kind: str,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store: ExecutionStore = (
+        FileSystemExecutionStore.create(tmp_path / "store")
+        if filesystem
+        else InMemoryExecutionStore()
+    )
+    supplied_error = (
+        ErrorObject(
+            message="rejected",
+            type="RejectedError",
+            data='{"reason":"test"}',
+            stack_trace=["original frame"],
+        )
+        if error_kind == "rich"
+        else ErrorObject.from_dict({})
+    )
+    expected_error = supplied_error if error_kind == "rich" else None
+    completed_operations: list[Operation] = []
+    original_complete = Execution.complete_callback_failure
+
+    def capture_live_completion(
+        execution: Execution,
+        callback_id: str,
+        error: ErrorObject,
+        now: datetime | None = None,
+    ) -> Operation:
+        # Capture the real API transition before the executor persists it.
+        # A filesystem round-trip must not hide missing boundary normalization.
+        operation = original_complete(execution, callback_id, error, now=now)
+        completed_operations.append(operation)
+        return operation
+
+    monkeypatch.setattr(Execution, "complete_callback_failure", capture_live_completion)
+    step_calls: list[str] = []
+    caught_errors: list[CallbackError] = []
+
+    @durable_step
+    def saved_step(_context: Any) -> str:
+        step_calls.append("saved")
+        return "saved"
+
+    def durable_handler(_event: Any, context: DurableContext) -> str:
+        saved = context.step(saved_step(), name="saved-step")
+        callback = context.create_callback(name="failed-callback")
+        try:
+            callback.result()
+        except CallbackError as error:
+            caught_errors.append(error)
+            return saved
+        raise AssertionError("Callback failure was not raised")
+
+    handler = durable_execution(durable_handler)
+    with DurableFunctionTestRunner(
+        handler=handler,
+        store=store,
+        skip_time=False,
+        execution_timeout=15,
+        poll_interval=0.01,
+    ) as runner:
+        arn = runner.run_async(input="{}")
+        callback_id = runner.wait_for_callback(arn, name="failed-callback", timeout=5)
+        deadline = time.monotonic() + 5
+        while not store.load(arn).invocation_completions:
+            assert time.monotonic() < deadline, "Initial invocation did not suspend"
+            time.sleep(0.01)
+        if error_kind == "omitted":
+            runner.send_callback_failure(callback_id)
+        else:
+            runner.send_callback_failure(callback_id, supplied_error)
+        result = runner.wait_for_result(arn, timeout=10)
+        detailed = runner.get_execution_history(arn, include_execution_data=True)
+        metadata_only = runner.get_execution_history(arn, include_execution_data=False)
+        saved_execution = store.load(arn)
+
+    assert result.status is InvocationStatus.SUCCEEDED
+    assert result.result == '"saved"'
+    assert step_calls == ["saved"]
+    assert len(saved_execution.invocation_completions) == 2
+    assert len(caught_errors) == 1
+    caught = caught_errors[0]
+    assert isinstance(caught, CallbackExternalError)
+    assert caught.message == (
+        expected_error.message if expected_error else "Callback failed"
+    )
+    assert caught.data == (expected_error.data if expected_error else None)
+    assert caught.stack_trace == (
+        expected_error.stack_trace if expected_error else None
+    )
+
+    assert len(completed_operations) == 1
+    live = completed_operations[0]
+    persisted = next(
+        op for op in saved_execution.operations if op.operation_id == live.operation_id
+    )
+    reloaded = Execution.from_json_dict(saved_execution.to_json_dict())
+    roundtripped = next(
+        op for op in reloaded.operations if op.operation_id == live.operation_id
+    )
+    for operation in (live, persisted, roundtripped):
+        assert operation.status is OperationStatus.FAILED
+        assert operation.callback_details is not None
+        assert operation.callback_details.error == expected_error
+
+    for history, include_data in ((detailed, True), (metadata_only, False)):
+        wire = history.to_dict()
+        failed = next(
+            event for event in wire["Events"] if event["EventType"] == "CallbackFailed"
+        )
+        expected: dict[str, Any] = {"Truncated": not include_data}
+        if include_data:
+            expected["Payload"] = expected_error.to_dict() if expected_error else {}
+        assert failed["CallbackFailedDetails"]["Error"] == expected
+        decoded = GetDurableExecutionHistoryResponse.from_dict(wire)
+        decoded_failed = next(
+            event for event in decoded.events if event.event_type == "CallbackFailed"
+        )
+        assert decoded_failed.callback_failed_details is not None
+        assert decoded_failed.callback_failed_details.error is not None
+        assert decoded_failed.callback_failed_details.error.to_dict() == expected
+        for snapshot in (history, decoded):
+            reconstructed = next(
+                op
+                for op in events_to_operations(snapshot.events)
+                if op.operation_id == live.operation_id
+            )
+            assert reconstructed.status is OperationStatus.FAILED
+            assert reconstructed.callback_details is not None
+            assert reconstructed.callback_details.error == (
+                expected_error if include_data else None
+            )