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/e2e/child_context_failure_test.py

@@ -0,0 +1,63 @@
+"""End-to-end child context failure handling through the test runner."""
+
+import json
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import StepConfig
+from aws_durable_execution_sdk_python.context import (
+    DurableContext,
+    durable_step,
+    durable_with_child_context,
+)
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.lambda_service import (
+    InvocationStatus,
+    OperationStatus,
+)
+from aws_durable_execution_sdk_python.retries import RetryPresets
+from aws_durable_execution_sdk_python.types import StepContext
+
+from aws_durable_execution_sdk_python_testing.runner import (
+    ContextOperation,
+    DurableFunctionTestResult,
+    DurableFunctionTestRunner,
+)
+
+
+def test_caught_child_context_failure_does_not_fail_root_execution() -> None:
+    @durable_step
+    def failing_step(step_context: StepContext) -> str:  # noqa: ARG001
+        msg = "Child step failed"
+        raise RuntimeError(msg)
+
+    @durable_with_child_context
+    def failing_child(ctx: DurableContext) -> str:
+        return ctx.step(
+            failing_step(),
+            config=StepConfig(retry_strategy=RetryPresets.none()),
+        )
+
+    @durable_step
+    def recovery_step(step_context: StepContext, value: str) -> str:  # noqa: ARG001
+        return value
+
+    @durable_execution
+    def handler(event: Any, context: DurableContext) -> str:  # noqa: ARG001
+        try:
+            context.run_in_child_context(failing_child(), name="failing-child")
+        except Exception:
+            pass
+
+        return context.step(recovery_step("handled"))
+
+    with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner:
+        result: DurableFunctionTestResult = runner.run(input="input str")
+
+    assert result.status is InvocationStatus.SUCCEEDED
+    assert result.result == json.dumps("handled")
+
+    child_op: ContextOperation = result.get_context("failing-child")
+    assert child_op.status == OperationStatus.FAILED
+    assert child_op.error is not None
+    assert child_op.error.message is not None
+    assert "Child step failed" in child_op.error.message

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_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
 ):