feat(serdes): Add itemSerdes and pass it when executing children

Changes:
- adds an itemSerdes field in map and parallel configs that will be
  passed onto run_in_child_context via Executor.execute_item and is used
  to serialize individual items for a batch.
- passes MapConfig.serdes and ParallelConfig.serdes when in the top
  level scope (top run_in_child_context) which will invoke the closure
  that then builds the executor.

Author: FullyTyped

src/aws_durable_execution_sdk_python/concurrency.py

@@ -566,6 +566,7 @@ def __init__(
         sub_type_iteration: OperationSubType,
         name_prefix: str,
         serdes: SerDes | None,
+        item_serdes: SerDes | None = None,
         summary_generator: SummaryGenerator | None = None,
     ):
         """Initialize ConcurrentExecutor.
@@ -604,6 +605,7 @@ def __init__(
         )
         self.executables_with_state: list[ExecutableWithState] = []
         self.serdes = serdes
+        self.item_serdes = item_serdes
 
     @abstractmethod
     def execute_item(
@@ -797,7 +799,7 @@ def execute_in_child_context(child_context: DurableContext) -> ResultType:
             execute_in_child_context,
             f"{self.name_prefix}{executable.index}",
             ChildConfig(
-                serdes=self.serdes,
+                serdes=self.item_serdes or self.serdes,
                 sub_type=self.sub_type_iteration,
                 summary_generator=self.summary_generator,
             ),

src/aws_durable_execution_sdk_python/config.py

@@ -125,9 +125,21 @@ class ParallelConfig:
             Default is CompletionConfig.all_successful() which requires all branches
             to succeed. Other options include first_successful() and all_completed().
 
-        serdes: Custom serialization/deserialization configuration for parallel results.
-            If None, uses the default serializer. This allows custom handling of
-            complex result types or optimization for large result sets.
+        serdes: Custom serialization/deserialization configuration for BatchResult.
+            Applied at the handler level to serialize the entire BatchResult object.
+            If None, uses the default JSON serializer for BatchResult.
+
+            Backward Compatibility: If only 'serdes' is provided (no item_serdes),
+            it will be used for both individual functions AND BatchResult serialization
+            to maintain existing behavior.
+
+        item_serdes: Custom serialization/deserialization configuration for individual functions.
+            Applied to each function's result as tasks complete in child contexts.
+            If None, uses the default JSON serializer for individual function results.
+
+            When both 'serdes' and 'item_serdes' are provided:
+            - item_serdes: Used for individual function results in child contexts
+            - serdes: Used for the entire BatchResult at handler level
 
         summary_generator: Function to generate compact summaries for large results (>256KB).
             When the serialized result exceeds CHECKPOINT_SIZE_LIMIT, this generator
@@ -150,6 +162,7 @@ class ParallelConfig:
         default_factory=CompletionConfig.all_successful
     )
     serdes: SerDes | None = None
+    item_serdes: SerDes | None = None
     summary_generator: SummaryGenerator | None = None
 
 
@@ -181,9 +194,21 @@ class ChildConfig(Generic[T]):
     matching the TypeScript ChildConfig interface behavior.
 
     Args:
-        serdes: Custom serialization/deserialization configuration for the child context data.
-            If None, uses the default serializer. This allows different serialization
-            strategies for child operations vs parent operations.
+        serdes: Custom serialization/deserialization configuration for BatchResult.
+            Applied at the handler level to serialize the entire BatchResult object.
+            If None, uses the default JSON serializer for BatchResult.
+
+            Backward Compatibility: If only 'serdes' is provided (no item_serdes),
+            it will be used for both individual items AND BatchResult serialization
+            to maintain existing behavior.
+
+        item_serdes: Custom serialization/deserialization configuration for individual items.
+            Applied to each item's result as tasks complete in child contexts.
+            If None, uses the default JSON serializer for individual items.
+
+            When both 'serdes' and 'item_serdes' are provided:
+            - item_serdes: Used for individual item results in child contexts
+            - serdes: Used for the entire BatchResult at handler level
 
         sub_type: Operation subtype identifier used for tracking and debugging.
             Examples: OperationSubType.MAP_ITERATION, OperationSubType.PARALLEL_BRANCH.
@@ -208,6 +233,7 @@ class ChildConfig(Generic[T]):
 
     # checkpoint_mode: CheckpointMode = CheckpointMode.CHECKPOINT_AT_START_AND_FINISH
     serdes: SerDes | None = None
+    item_serdes: SerDes | None = None
     sub_type: OperationSubType | None = None
     summary_generator: SummaryGenerator | None = None
 
@@ -273,9 +299,21 @@ class MapConfig:
             Default allows any number of failures. Use CompletionConfig.all_successful()
             to require all items to succeed.
 
-        serdes: Custom serialization/deserialization configuration for map results.
-            If None, uses the default serializer. This allows custom handling of
-            complex item types or optimization for large result collections.
+        serdes: Custom serialization/deserialization configuration for BatchResult.
+            Applied at the handler level to serialize the entire BatchResult object.
+            If None, uses the default JSON serializer for BatchResult.
+
+            Backward Compatibility: If only 'serdes' is provided (no item_serdes),
+            it will be used for both individual items AND BatchResult serialization
+            to maintain existing behavior.
+
+        item_serdes: Custom serialization/deserialization configuration for individual items.
+            Applied to each item's result as tasks complete in child contexts.
+            If None, uses the default JSON serializer for individual items.
+
+            When both 'serdes' and 'item_serdes' are provided:
+            - item_serdes: Used for individual item results in child contexts
+            - serdes: Used for the entire BatchResult at handler level
 
         summary_generator: Function to generate compact summaries for large results (>256KB).
             When the serialized result exceeds CHECKPOINT_SIZE_LIMIT, this generator
@@ -298,6 +336,7 @@ class MapConfig:
     item_batcher: ItemBatcher = field(default_factory=ItemBatcher)
     completion_config: CompletionConfig = field(default_factory=CompletionConfig)
     serdes: SerDes | None = None
+    item_serdes: SerDes | None = None
     summary_generator: SummaryGenerator | None = None
 
 

src/aws_durable_execution_sdk_python/context.py

@@ -303,7 +303,7 @@ def map(
         """Execute a callable for each item in parallel."""
         map_name: str | None = self._resolve_step_name(name, func)
 
-        def map_in_child_context(child_context):
+        def map_in_child_context(child_context) -> BatchResult[R]:
             return map_handler(
                 items=inputs,
                 func=func,
@@ -315,7 +315,10 @@ def map_in_child_context(child_context):
         return self.run_in_child_context(
             func=map_in_child_context,
             name=map_name,
-            config=ChildConfig(sub_type=OperationSubType.MAP),
+            config=ChildConfig(
+                sub_type=OperationSubType.MAP,
+                serdes=config.serdes if config is not None else None,
+            ),
         )
 
     def parallel(
@@ -337,7 +340,10 @@ def parallel_in_child_context(child_context) -> BatchResult[T]:
         return self.run_in_child_context(
             func=parallel_in_child_context,
             name=name,
-            config=ChildConfig(sub_type=OperationSubType.PARALLEL),
+            config=ChildConfig(
+                sub_type=OperationSubType.PARALLEL,
+                serdes=config.serdes if config is not None else None,
+            ),
         )
 
     def run_in_child_context(

src/aws_durable_execution_sdk_python/operation/map.py

@@ -21,7 +21,6 @@
     from aws_durable_execution_sdk_python.state import ExecutionState
     from aws_durable_execution_sdk_python.types import DurableContext, SummaryGenerator
 
-
 logger = logging.getLogger(__name__)
 
 # Input item type
@@ -42,6 +41,7 @@ def __init__(
         name_prefix: str,
         serdes: SerDes | None,
         summary_generator: SummaryGenerator | None = None,
+        item_serdes: SerDes | None = None,
     ):
         super().__init__(
             executables=executables,
@@ -52,6 +52,7 @@ def __init__(
             name_prefix=name_prefix,
             serdes=serdes,
             summary_generator=summary_generator,
+            item_serdes=item_serdes,
         )
         self.items = items
 

src/aws_durable_execution_sdk_python/operation/parallel.py

@@ -35,6 +35,7 @@ def __init__(
         name_prefix: str,
         serdes: SerDes | None,
         summary_generator: SummaryGenerator | None = None,
+        item_serdes: SerDes | None = None,
     ):
         super().__init__(
             executables=executables,
@@ -45,6 +46,7 @@ def __init__(
             name_prefix=name_prefix,
             serdes=serdes,
             summary_generator=summary_generator,
+            item_serdes=item_serdes,
         )
 
     @classmethod
@@ -98,7 +100,7 @@ def parallel_handler(
 
 
 class ParallelSummaryGenerator:
-    def __call__(self, result: BatchResult[R]) -> str:
+    def __call__(self, result: BatchResult) -> str:
         fields = {
             "totalCount": result.total_count,
             "successCount": result.success_count,

tests/operation/child_test.py

@@ -1,6 +1,7 @@
 """Unit tests for child handler."""
 
 import json
+from typing import cast
 from unittest.mock import Mock
 
 import pytest
@@ -16,6 +17,7 @@
 )
 from aws_durable_execution_sdk_python.operation.child import child_handler
 from aws_durable_execution_sdk_python.state import ExecutionState
+from aws_durable_execution_sdk_python.types import SummaryGenerator
 from tests.serdes_test import CustomDictSerDes
 
 
@@ -383,7 +385,9 @@ def test_child_handler_large_payload_with_summary_generator() -> None:
     def my_summary(result: str) -> str:
         return "summary"
 
-    child_config: ChildConfig = ChildConfig[str](summary_generator=my_summary)
+    child_config: ChildConfig = ChildConfig[str](
+        summary_generator=cast(SummaryGenerator, my_summary)
+    )
 
     actual_result = child_handler(
         mock_callable,