fix(testing): handle workflow failure states correctly

Author: hln33

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

@@ -406,7 +406,9 @@ def complete_success(self, result: str | None, now: datetime | None = None) -> N
         self.close_status = ExecutionStatus.SUCCEEDED
         self._end_execution(OperationStatus.SUCCEEDED, now)
 
-    def complete_fail(self, error: ErrorObject, now: datetime | None = None) -> None:
+    def complete_fail(
+        self, error: ErrorObject | None, now: datetime | None = None
+    ) -> None:
         """Complete execution with failure (DecisionType.FAIL_WORKFLOW_EXECUTION)."""
         self.result = DurableExecutionInvocationOutput(
             status=InvocationStatus.FAILED, error=error

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

@@ -1285,9 +1285,7 @@ def _validate_invocation_response_and_store(
                     )
                     raise InvalidParameterValueException(msg_failed_result)
                 logger.info("[%s] Execution failed", execution_arn)
-                self._complete_workflow(
-                    execution_arn, result=None, error=response.error
-                )
+                self._fail_workflow(execution_arn, response.error)
 
             case InvocationStatus.SUCCEEDED:
                 if response.error is not None:
@@ -1591,7 +1589,7 @@ def _complete_workflow(
         else:
             self.complete_execution(execution_arn, result)
 
-    def _fail_workflow(self, execution_arn: str, error: ErrorObject):
+    def _fail_workflow(self, execution_arn: str, error: ErrorObject | None):
         """Fail workflow with terminal state validation."""
         execution = self._store.load(execution_arn)
 
@@ -1671,8 +1669,8 @@ def complete_execution(self, execution_arn: str, result: str | None = None) -> N
             raise IllegalStateException(msg)
         self._complete_events(execution_arn=execution_arn)
 
-    def fail_execution(self, execution_arn: str, error: ErrorObject) -> None:
-        """Fail execution with error (FAIL_WORKFLOW_EXECUTION decision)."""
+    def fail_execution(self, execution_arn: str, error: ErrorObject | None) -> None:
+        """Fail execution with optional error (FAIL_WORKFLOW_EXECUTION decision)."""
         logger.error("[%s] Completing execution with error: %s", execution_arn, error)
         execution: Execution = self._store.load(execution_arn=execution_arn)
         execution.complete_fail(error=error, now=self._clock.now())
@@ -1688,7 +1686,7 @@ def on_completed(self, execution_arn: str, result: str | None = None) -> None:
         """Complete execution successfully. Observer method triggered by notifier."""
         self.complete_execution(execution_arn, result)
 
-    def on_failed(self, execution_arn: str, error: ErrorObject) -> None:
+    def on_failed(self, execution_arn: str, error: ErrorObject | None) -> None:
         """Fail execution. Observer method triggered by notifier."""
         self.fail_execution(execution_arn, error)
 

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

@@ -158,6 +158,41 @@ def test_create_execution_failed():
     assert event.execution_failed_details.error.payload.message == "Execution failed"
 
 
+def test_create_execution_failed_without_error_payload():
+    from aws_durable_execution_sdk_python.execution import (
+        DurableExecutionInvocationOutput,
+        InvocationStatus,
+    )
+
+    operation = create_mock_operation("op-1", status=OperationStatus.FAILED)
+    operation.end_timestamp = datetime.now(UTC)
+
+    error_result = DurableExecutionInvocationOutput(
+        status=InvocationStatus.FAILED,
+        error=None,
+    )
+    context = EventCreationContext.create(
+        operation=operation,
+        event_id=3,
+        durable_execution_arn="arn:test",
+        start_input=StartDurableExecutionInput(
+            account_id="123",
+            function_name="test",
+            function_qualifier="$LATEST",
+            execution_name="test",
+            execution_timeout_seconds=300,
+            execution_retention_period_days=7,
+        ),
+        result=error_result,
+        include_execution_data=True,
+    )
+    event = Event.create_execution_event(context)
+
+    assert event.event_type == "ExecutionFailed"
+    assert event.execution_failed_details.error is not None
+    assert event.execution_failed_details.error.payload is None
+
+
 def test_create_execution_timed_out():
     from aws_durable_execution_sdk_python.execution import (
         DurableExecutionInvocationOutput,

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

@@ -507,6 +507,26 @@ def test_complete_fail():
     assert execution.result.error == error
 
 
+def test_complete_fail_without_error():
+    """Test complete_fail preserves a missing error payload."""
+    start_input = StartDurableExecutionInput(
+        account_id="123456789012",
+        function_name="test-function",
+        function_qualifier="$LATEST",
+        execution_name="test-execution",
+        execution_timeout_seconds=300,
+        execution_retention_period_days=7,
+        invocation_id="test-invocation-id",
+    )
+    execution = Execution("test-arn", start_input, [Mock()])
+
+    execution.complete_fail(None)
+
+    assert execution.is_complete is True
+    assert execution.result.status == InvocationStatus.FAILED
+    assert execution.result.error is None
+
+
 def test_find_operation_exists():
     """Test find_operation method when operation exists."""
     start_input = StartDurableExecutionInput(

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

@@ -21,6 +21,8 @@
 * Gate-release: after a sequence of transient invocation failures,
   the observed InvocationState is never INVOKING.
 * Idempotent replay byte-equivalence.
+* Child context failures remain operation-local and do not force a
+  successful root invocation to fail.
 """
 
 from __future__ import annotations
@@ -52,6 +54,7 @@
 )
 from aws_durable_execution_sdk_python_testing.execution import (
     Execution,
+    ExecutionStatus,
     OperationPaginatorState,
 )
 from aws_durable_execution_sdk_python_testing.executor import Executor, InvocationState
@@ -927,6 +930,58 @@ def test_checkpoint_returns_full_delta_in_one_response():
     assert [op.operation_id for op in r2.new_execution_state.operations] == []
 
 
+def test_context_failed_checkpoint_does_not_force_root_failure():
+    """A caught child context failure must not force root execution failure."""
+    executor, store, arn, token = _make_executor()
+    error = ErrorObject.from_message("context failed")
+
+    executor.checkpoint_execution(
+        execution_arn=arn,
+        checkpoint_token=token,
+        updates=[
+            OperationUpdate(
+                operation_id="context-1",
+                operation_type=OperationType.CONTEXT,
+                action=OperationAction.START,
+                name="context-1",
+            ),
+            OperationUpdate(
+                operation_id="context-1",
+                operation_type=OperationType.CONTEXT,
+                action=OperationAction.FAIL,
+                name="context-1",
+                error=error,
+            ),
+        ],
+    )
+
+    checkpointed_execution = store.load(arn)
+    _, context_op = checkpointed_execution.find_operation("context-1")
+    assert context_op.operation_type == OperationType.CONTEXT
+    assert context_op.status == OperationStatus.FAILED
+    assert context_op.context_details is not None
+    assert context_op.context_details.error == error
+
+    response = DurableExecutionInvocationOutput(
+        status=InvocationStatus.SUCCEEDED, result="handled"
+    )
+    executor._validate_invocation_response_and_store(  # noqa: SLF001
+        arn, response, checkpointed_execution
+    )
+
+    final_execution = store.load(arn)
+    assert final_execution.is_complete is True
+    assert final_execution.close_status == ExecutionStatus.SUCCEEDED
+    assert final_execution.result is not None
+    assert final_execution.result.status == InvocationStatus.SUCCEEDED
+    assert final_execution.result.result == "handled"
+
+    _, context_op = final_execution.find_operation("context-1")
+    assert context_op.status == OperationStatus.FAILED
+    assert context_op.context_details is not None
+    assert context_op.context_details.error == error
+
+
 def test_validate_invocation_response_raises_on_already_complete():
     """Covers the defensive is_complete check in
     _validate_invocation_response_and_store. Direct invocation with a

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

@@ -310,6 +310,42 @@ def test_should_complete_workflow_with_error_when_invocation_fails(
         mock_fail.assert_called_once_with("test-arn", failed_response.error)
 
 
+def test_validate_invocation_response_failed_without_error_still_fails():
+    """FAILED without an error must fail (not succeed), preserving the null error."""
+    
+    store = InMemoryExecutionStore()
+    executor = Executor(store, Mock(), Mock(), Mock())
+
+    start_input = StartDurableExecutionInput(
+        account_id="123456789012",
+        function_name="test-function",
+        function_qualifier="$LATEST",
+        execution_name="test-execution",
+        execution_timeout_seconds=300,
+        execution_retention_period_days=7,
+        invocation_id="test-invocation-id",
+    )
+    execution = Execution.new(start_input)
+    execution.start()
+    store.save(execution)
+
+    response = DurableExecutionInvocationOutput(
+        status=InvocationStatus.FAILED, error=None
+    )
+
+    executor._validate_invocation_response_and_store(  # noqa: SLF001
+        execution.durable_execution_arn, response, execution
+    )
+
+    stored = store.load(execution.durable_execution_arn)
+    assert stored.is_complete is True
+    assert stored.close_status is not None
+    assert stored.close_status.value == "FAILED"
+    assert stored.result is not None
+    assert stored.result.status == InvocationStatus.FAILED
+    assert stored.result.error is None
+
+
 def test_should_complete_workflow_with_result_when_invocation_succeeds(
     executor, mock_store, mock_scheduler, mock_invoker, start_input
 ):