fix: keep empty string payloads when serializing

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

@@ -470,7 +470,7 @@ def to_dict(self) -> MutableMapping[str, Any]:
             result["Name"] = self.name
         if self.sub_type:
             result["SubType"] = self.sub_type.value
-        if self.payload:
+        if self.payload is not None:
             result["Payload"] = self.payload
         if self.error:
             result["Error"] = self.error.to_dict()

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

@@ -743,6 +743,19 @@ def test_operation_update_create_step_succeed():
     assert update.sub_type is OperationSubType.STEP
 
 
+def test_operation_update_to_dict_includes_empty_payload():
+    """Test OperationUpdate.to_dict includes an empty-string payload."""
+    update = OperationUpdate.create_step_succeed(
+        OperationIdentifier("step1", OperationSubType.STEP, None, "empty_result"),
+        payload="",
+    )
+
+    assert update.payload == ""
+    result = update.to_dict()
+    assert "Payload" in result
+    assert result["Payload"] == ""
+
+
 def test_operation_update_factory_methods():
     """Test all OperationUpdate factory methods."""
     error = ErrorObject(

packages/aws-durable-execution-sdk-python/tests/operation/step_test.py

@@ -31,7 +31,11 @@
 from aws_durable_execution_sdk_python.logger import Logger
 from aws_durable_execution_sdk_python.operation.step import StepOperationExecutor
 from aws_durable_execution_sdk_python.retries import RetryDecision
-from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext
+from aws_durable_execution_sdk_python.serdes import (
+    PassThroughSerDes,
+    SerDes,
+    SerDesContext,
+)
 from aws_durable_execution_sdk_python.state import CheckpointedResult, ExecutionState
 from aws_durable_execution_sdk_python.types import StepContext
 from tests.serdes_test import CustomDictSerDes
@@ -774,6 +778,66 @@ def deserialize(self, _data: str, _ctx: SerDesContext) -> Any:
     assert result is None
 
 
+def test_step_handler_empty_string_result_first_run():
+    """Test step_handler checkpoints an empty-string result as an empty payload."""
+    mock_state = Mock(spec=ExecutionState)
+    mock_state.get_checkpoint_result.return_value = (
+        CheckpointedResult.create_not_found()
+    )
+    mock_state.durable_execution_arn = "test_arn"
+    mock_callable = Mock(return_value="")
+    mock_state.wrap_user_function.return_value = mock_callable
+    mock_logger = Mock(spec=Logger)
+    mock_logger.with_log_info.return_value = mock_logger
+
+    config = StepConfig(
+        step_semantics=StepSemantics.AT_LEAST_ONCE_PER_RETRY,
+        serdes=PassThroughSerDes(),
+    )
+    result = step_handler(
+        mock_callable,
+        mock_state,
+        OperationIdentifier("step1", OperationSubType.STEP, None, "test_step"),
+        config,
+        mock_logger,
+    )
+
+    assert result == ""
+
+    success_call = mock_state.create_checkpoint.call_args_list[1]
+    success_operation = success_call[1]["operation_update"]
+    assert success_operation.payload == ""
+    assert "Payload" in success_operation.to_dict()
+
+
+def test_step_handler_already_succeeded_empty_string_result():
+    """Test step_handler when operation succeeded with an empty-string result."""
+    mock_state = Mock(spec=ExecutionState)
+    mock_state.durable_execution_arn = "test_arn"
+    operation = Operation(
+        operation_id="step1",
+        operation_type=OperationType.STEP,
+        status=OperationStatus.SUCCEEDED,
+        step_details=StepDetails(result=""),
+    )
+    mock_result = CheckpointedResult.create_from_operation(operation)
+    mock_state.get_checkpoint_result.return_value = mock_result
+
+    mock_callable = Mock()
+    mock_logger = Mock(spec=Logger)
+
+    result = step_handler(
+        mock_callable,
+        mock_state,
+        OperationIdentifier("step1", OperationSubType.STEP, None, "test_step"),
+        StepConfig(serdes=PassThroughSerDes()),
+        mock_logger,
+    )
+
+    assert result == ""
+    mock_callable.assert_not_called()
+
+
 # Tests for immediate response handling