fix(sdk): filter batch items on status alone

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py

@@ -32,6 +32,7 @@
     NestingType,
 )
 from aws_durable_execution_sdk_python.exceptions import (
+    ChildContextError,
     DurableOperationError,
     ExecutionError,
     InvalidStateError,
@@ -89,6 +90,21 @@ def _branch_error_object(err: Exception) -> ErrorObject:
     return ErrorObject.from_exception(err)
 
 
+def _failed_checkpoint_error_object(checkpoint: CheckpointedResult) -> ErrorObject:
+    """Return the error of a branch whose child context checkpoint FAILED.
+
+    The checkpoint error is optional. Without one, record the generic
+    ChildContextError that replaying the child context raises, so every
+    FAILED item carries an error.
+    """
+    if checkpoint.error is not None:
+        return checkpoint.error
+    try:
+        checkpoint.raise_operation_error(ChildContextError)
+    except ChildContextError as e:
+        return _branch_error_object(e)
+
+
 @dataclass
 class ResumeWave(Generic[CallableType, ResultType]):
     """Branches suspended until one time, and the refresh that resumes them.
@@ -796,7 +812,9 @@ def _replay_terminal_item(
             return BatchItem(executable.index, BatchItemStatus.SUCCEEDED, result)
         if checkpoint.is_failed():
             return BatchItem(
-                executable.index, BatchItemStatus.FAILED, error=checkpoint.error
+                executable.index,
+                BatchItemStatus.FAILED,
+                error=_failed_checkpoint_error_object(checkpoint),
             )
         if self.nesting_type is NestingType.FLAT:
             try:
@@ -849,7 +867,7 @@ def _replay_from_checkpoints(
                 )
 
             elif checkpoint.is_failed():
-                error = checkpoint.error
+                error = _failed_checkpoint_error_object(checkpoint)
                 status = BatchItemStatus.FAILED
             else:
                 status = BatchItemStatus.STARTED

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py

@@ -7,7 +7,7 @@
 from collections import Counter
 from dataclasses import dataclass
 from enum import Enum
-from typing import TYPE_CHECKING, Generic, TypeVar
+from typing import TYPE_CHECKING, Generic, TypeVar, cast
 
 from aws_durable_execution_sdk_python.exceptions import (
     ChildContextError,
@@ -42,6 +42,26 @@
 
 # region Result models
 
+_MISSING_FAILED_ITEM_ERROR_MESSAGE = (
+    "Unknown error. No ErrorObject exists on the Checkpoint Operation."
+)
+
+
+def _generic_failed_item_error() -> ErrorObject:
+    """ErrorObject for a FAILED batch item that carried no checkpoint error.
+
+    Matches ``_failed_checkpoint_error_object`` when the child-context
+    checkpoint has no ErrorObject: the same generic ChildContextError
+    replaying that child would raise.
+    """
+    err = ChildContextError(message=_MISSING_FAILED_ITEM_ERROR_MESSAGE)
+    return ErrorObject(
+        message=err.message,
+        type=err.error_type,
+        data=err.data,
+        stack_trace=err.stack_trace,
+    )
+
 
 class CompletionReason(Enum):
     ALL_COMPLETED = "ALL_COMPLETED"
@@ -401,6 +421,10 @@ class BatchItem(Generic[R]):
     result: R | None = None
     error: ErrorObject | None = None
 
+    def __post_init__(self) -> None:
+        if self.status is BatchItemStatus.FAILED and self.error is None:
+            object.__setattr__(self, "error", _generic_failed_item_error())
+
     def to_dict(self) -> dict:
         return {
             "index": self.index,
@@ -489,18 +513,10 @@ def to_dict(self) -> dict:
         }
 
     def succeeded(self) -> list[BatchItem[R]]:
-        return [
-            item
-            for item in self.all
-            if item.status is BatchItemStatus.SUCCEEDED and item.result is not None
-        ]
+        return [item for item in self.all if item.status is BatchItemStatus.SUCCEEDED]
 
     def failed(self) -> list[BatchItem[R]]:
-        return [
-            item
-            for item in self.all
-            if item.status is BatchItemStatus.FAILED and item.error is not None
-        ]
+        return [item for item in self.all if item.status is BatchItemStatus.FAILED]
 
     def started(self) -> list[BatchItem[R]]:
         return [item for item in self.all if item.status is BatchItemStatus.STARTED]
@@ -533,17 +549,13 @@ def throw_if_error(self) -> None:
             raise BatchCompletionError(self.completion_reason)
 
     def get_results(self) -> list[R]:
-        return [
-            item.result
-            for item in self.all
-            if item.status is BatchItemStatus.SUCCEEDED and item.result is not None
-        ]
+        # None is a valid result for a branch that returns nothing.
+        return [cast("R", item.result) for item in self.succeeded()]
 
     def get_errors(self) -> list[ErrorObject]:
         return [
-            item.error
-            for item in self.all
-            if item.status is BatchItemStatus.FAILED and item.error is not None
+            item.error if item.error is not None else _generic_failed_item_error()
+            for item in self.failed()
         ]
 
     @property

packages/aws-durable-execution-sdk-python/tests/concurrency_test.py

@@ -61,6 +61,7 @@
     SuspendExecution,
     TimedSuspendExecution,
 )
+from aws_durable_execution_sdk_python.serdes import ExtendedTypeSerDes
 from aws_durable_execution_sdk_python.lambda_service import (
     DurableServiceClient,
     ErrorObject,
@@ -310,6 +311,23 @@ def test_batch_result_get_errors():
     assert error2 in errors
 
 
+def test_batch_result_keeps_succeeded_items_with_none_result():
+    """A branch that returns None still counts as succeeded."""
+    items = [
+        BatchItem(0, BatchItemStatus.SUCCEEDED, None),
+        BatchItem(1, BatchItemStatus.SUCCEEDED, 10),
+        BatchItem(
+            2, BatchItemStatus.FAILED, error=ErrorObject("msg", "Error", None, None)
+        ),
+        BatchItem(3, BatchItemStatus.SUCCEEDED, None),
+    ]
+    result = BatchResult(items, CompletionReason.ALL_COMPLETED)
+
+    assert [item.index for item in result.succeeded()] == [0, 1, 3]
+    assert len(result.succeeded()) == result.success_count
+    assert result.get_results() == [None, 10, None]
+
+
 def test_batch_result_counts():
     """Test BatchResult count properties."""
     items = [
@@ -596,18 +614,74 @@ def test_func():
 
 
 def test_batch_result_failed_with_none_error():
-    """Test BatchResult failed method filters out None errors."""
+    """FAILED items without an error normalize to ChildContextError.
+
+    Construction and deserialization both fill the same generic error the
+    executor records for a FAILED child-context checkpoint with no
+    ErrorObject, so get_errors() never returns None.
+    """
+    child_context_type = (
+        "aws_durable_execution_sdk_python.exceptions.ChildContextError"
+    )
     items = [
-        BatchItem(0, BatchItemStatus.FAILED, error=None),  # Should be filtered out
+        BatchItem(0, BatchItemStatus.FAILED, error=None),
         BatchItem(
             1, BatchItemStatus.FAILED, error=ErrorObject("msg", "Error", None, None)
         ),
     ]
     result = BatchResult(items, CompletionReason.ALL_COMPLETED)
 
     failed = result.failed()
-    assert len(failed) == 1
+    assert len(failed) == result.failure_count == 2
     assert failed[0].error is not None
+    assert failed[0].error.type == child_context_type
+    assert failed[0].error.message == (
+        "Unknown error. No ErrorObject exists on the Checkpoint Operation."
+    )
+
+    errors = result.get_errors()
+    assert len(errors) == 2
+    assert errors[0].type == child_context_type
+    assert errors[1].type == "Error"
+
+    # Wire payload may still omit error; from_dict must normalize too.
+    wire = {
+        "all": [
+            {"index": 0, "status": "FAILED", "result": None, "error": None},
+            {
+                "index": 1,
+                "status": "FAILED",
+                "result": None,
+                "error": ErrorObject("msg", "Error", None, None).to_dict(),
+            },
+        ],
+        "completionReason": "ALL_COMPLETED",
+    }
+    round_tripped = BatchResult.from_dict(wire)
+    round_trip_errors = round_tripped.get_errors()
+    assert len(round_trip_errors) == 2
+    assert round_trip_errors[0].type == child_context_type
+    assert all(error is not None for error in round_trip_errors)
+    assert round_trip_errors[0].type is not None
+    # Accessing error.type must not raise after deserialization.
+    assert [error.type for error in round_trip_errors] == [
+        child_context_type,
+        "Error",
+    ]
+
+    # Same gap after ExtendedTypeSerDes round-trip of a wire payload with
+    # error=null (the path that previously yielded AttributeError on .type).
+    serdes_payload = (
+        '{"t":"br","v":{"all":{"t":"l","v":[{"t":"m","v":{"index":{"t":"i","v":0},'
+        '"status":{"t":"s","v":"FAILED"},"result":{"t":"n","v":null},'
+        '"error":{"t":"n","v":null}}}]},"completionReason":{"t":"s","v":"ALL_COMPLETED"}}}'
+    )
+    serdes_result = ExtendedTypeSerDes().deserialize(serdes_payload)
+    assert isinstance(serdes_result, BatchResult)
+    serdes_errors = serdes_result.get_errors()
+    assert len(serdes_errors) == 1
+    assert serdes_errors[0].type == child_context_type
+
 
 
 def test_concurrent_executor_nesting_type_parameter():
@@ -1035,6 +1109,58 @@ def test_flat_replay_rejects_failed_nested_container_checkpoint(replay_path):
     executor_context.create_child_context.assert_not_called()
 
 
+@pytest.mark.parametrize("replay_path", ["recorded-terminal", "fallback"])
+def test_replay_failed_checkpoint_without_error_records_child_context_error(
+    replay_path,
+):
+    """A FAILED branch checkpoint may carry no error; the item still gets one."""
+    executor = _RecordingExecutor(
+        executables=[Executable(0, lambda: "must-not-run")],
+        max_concurrency=1,
+        completion_config=CompletionConfig(tolerated_failure_count=1),
+        sub_type_top=OperationSubType.PARALLEL,
+        sub_type_iteration=OperationSubType.PARALLEL_BRANCH,
+        name_prefix="parallel-branch-",
+        serdes=None,
+        nesting_type=NestingType.NESTED,
+        operation_id_namespace=_StubNamespace(),
+    )
+    executor_context = Mock()
+    executor_context._parent_id = "parallel-op"  # noqa: SLF001
+    execution_state = Mock()
+    execution_state.get_checkpoint_result.return_value = (
+        CheckpointedResult.create_from_operation(
+            Operation(
+                operation_id="op_0",
+                operation_type=OperationType.CONTEXT,
+                status=OperationStatus.FAILED,
+                parent_id="parallel-op",
+                sub_type=OperationSubType.PARALLEL_BRANCH,
+                name="parallel-branch-0",
+            )
+        )
+    )
+
+    if replay_path == "recorded-terminal":
+        item = executor._replay_terminal_item(  # noqa: SLF001
+            execution_state, executor_context, executor.executables[0]
+        )
+        result = BatchResult([item], CompletionReason.ALL_COMPLETED)
+    else:
+        result = executor._replay_from_checkpoints(  # noqa: SLF001
+            execution_state, executor_context
+        )
+
+    assert [item.index for item in result.failed()] == [0]
+    errors = result.get_errors()
+    assert len(errors) == 1
+    assert errors[0].type == (
+        "aws_durable_execution_sdk_python.exceptions.ChildContextError"
+    )
+    with pytest.raises(ChildContextError, match="No ErrorObject exists"):
+        result.throw_if_error()
+
+
 @pytest.mark.parametrize(
     ("nesting_type", "checkpoint_name", "mismatch"),
     [
@@ -4007,6 +4133,22 @@ def suspend():
     assert replayed.all[2].result == "replayed_2"
 
 
+def test_execute_keeps_branches_that_return_none():
+    """Branches returning None appear in succeeded() and get_results()."""
+    executables = [
+        Executable(index, partial(lambda value: value, value))
+        for index, value in enumerate([None, 10, None, 30])
+    ]
+    executor = _make_executor(executables, max_concurrency=2)
+    execution_state, executor_context = _make_executor_mocks()
+
+    result: BatchResult = executor.execute(execution_state, executor_context)
+
+    assert result.success_count == 4
+    assert len(result.succeeded()) == 4
+    assert result.get_results() == [None, 10, None, 30]
+
+
 def test_replay_without_record_falls_back_to_checkpoint_derivation():
     """No record: every executable is represented, old-style."""
     executables = [Executable(i, lambda: "x") for i in range(3)]