fix: exit earily on empty list in map and parallel

Fix crash when calling context.map() or context.parallel() with empty
list. Previously, empty executables caused ThreadPoolExecutor to be
initialized with max_workers=0, raising ValueError.

Add early return in ConcurrentExecutor.execute() to immediately return
empty BatchResult when no items need processing. This avoids creating
ThreadPoolExecutor with invalid worker count and matches expected
behavior where empty input produces empty output.

Add regression test to verify empty list handling works correctly.

closes aws#323

Author: yaythomas

src/aws_durable_execution_sdk_python/concurrency/executor.py

@@ -30,6 +30,7 @@
 from aws_durable_execution_sdk_python.lambda_service import ErrorObject
 from aws_durable_execution_sdk_python.operation.child import child_handler
 
+
 if TYPE_CHECKING:
     from collections.abc import Callable
 
@@ -196,6 +197,11 @@ def execute(
             "▶️ Executing concurrent operation, items: %d", len(self.executables)
         )
 
+        # Early return for empty executables
+        if not self.executables:
+            logger.debug("No items to execute, returning empty result")
+            return self._create_result()
+
         max_workers = self.max_concurrency or len(self.executables)
 
         self.executables_with_state = [

tests/operation/map_test.py

@@ -1110,3 +1110,38 @@ def create_id(self, i):
         assert parent_call[1]["serdes"] is custom_serdes
         assert isinstance(parent_call[1]["value"], BatchResult)
         assert parent_call[1]["value"] is result
+
+
+def test_map_with_empty_list_should_exit_early():
+    """Test that map with empty list completes without crashing."""
+    items = []
+
+    def map_func(ctx, item, idx, items):
+        return f"processed_{item}"
+
+    mock_state = Mock()
+    mock_state.durable_execution_arn = "arn:test"
+
+    parent_checkpoint = Mock()
+    parent_checkpoint.is_succeeded.return_value = False
+    parent_checkpoint.is_failed.return_value = False
+    parent_checkpoint.is_existent.return_value = False
+
+    mock_state.get_checkpoint_result = Mock(return_value=parent_checkpoint)
+    mock_state.create_checkpoint = Mock()
+
+    context = create_test_context(state=mock_state)
+
+    # This should complete immediately without crashing
+    result = context.map(
+        items,
+        map_func,
+        name="EmptyMap",
+    )
+
+    # Should return empty BatchResult
+    assert isinstance(result, BatchResult)
+    assert len(result.all) == 0
+    assert result.total_count == 0
+    assert result.success_count == 0
+    assert result.failure_count == 0