fix: fix callback serdes

- Add a new passthrough serdes
- If the customer does not provide customized serdes for callback handler, use passthrough serdes for callback result because they are not created by sdk, instead, they are created by backend with customer data.

Author: Alex Wang

src/aws_durable_execution_sdk_python/context.py

@@ -36,7 +36,11 @@
 from aws_durable_execution_sdk_python.operation.wait_for_condition import (
     wait_for_condition_handler,
 )
-from aws_durable_execution_sdk_python.serdes import SerDes, deserialize
+from aws_durable_execution_sdk_python.serdes import (
+    PassThroughSerDes,
+    SerDes,
+    deserialize,
+)
 from aws_durable_execution_sdk_python.state import ExecutionState  # noqa: TCH001
 from aws_durable_execution_sdk_python.threading import OrderedCounter
 from aws_durable_execution_sdk_python.types import (
@@ -66,6 +70,8 @@
 
 logger = logging.getLogger(__name__)
 
+PASS_THROUGH_SERDES: SerDes[Any] = PassThroughSerDes()
+
 
 def durable_step(
     func: Callable[Concatenate[StepContext, Params], T],
@@ -144,7 +150,7 @@ def result(self) -> T | None:
                 return None  # type: ignore
 
             return deserialize(
-                serdes=self.serdes,
+                serdes=self.serdes if self.serdes is not None else PASS_THROUGH_SERDES,
                 data=checkpointed_result.result,
                 operation_id=self.operation_id,
                 durable_execution_arn=self.state.durable_execution_arn,

src/aws_durable_execution_sdk_python/serdes.py

@@ -372,6 +372,14 @@ def is_primitive(obj: Any) -> bool:
         return False
 
 
+class PassThroughSerDes(SerDes[T]):
+    def serialize(self, value: T, _: SerDesContext) -> str:  # noqa: PLR6301
+        return value  # type: ignore
+
+    def deserialize(self, data: str, _: SerDesContext) -> T:  # noqa: PLR6301
+        return data  # type: ignore
+
+
 class JsonSerDes(SerDes[T]):
     def serialize(self, value: T, _: SerDesContext) -> str:  # noqa: PLR6301
         return json.dumps(value)

tests/context_test.py

@@ -75,6 +75,28 @@ def test_callback_result_succeeded():
     callback = Callback("callback1", "op1", mock_state)
     result = callback.result()
 
+    assert result == '"success_result"'
+    mock_state.get_checkpoint_result.assert_called_once_with("op1")
+
+
+def test_callback_result_succeeded_with_plain_str():
+    """Test Callback.result() when operation succeeded."""
+    mock_state = Mock(spec=ExecutionState)
+    mock_state.durable_execution_arn = "test_arn"
+    operation = Operation(
+        operation_id="op1",
+        operation_type=OperationType.CALLBACK,
+        status=OperationStatus.SUCCEEDED,
+        callback_details=CallbackDetails(
+            callback_id="callback1", result="success_result"
+        ),
+    )
+    mock_result = CheckpointedResult.create_from_operation(operation)
+    mock_state.get_checkpoint_result.return_value = mock_result
+
+    callback = Callback("callback1", "op1", mock_state)
+    result = callback.result()
+
     assert result == "success_result"
     mock_state.get_checkpoint_result.assert_called_once_with("op1")
 

tests/serdes_test.py

@@ -28,6 +28,7 @@
     EncodedValue,
     ExtendedTypeSerDes,
     JsonSerDes,
+    PassThroughSerDes,
     PrimitiveCodec,
     SerDes,
     SerDesContext,
@@ -737,6 +738,18 @@ def test_extended_serdes_errors():
 # endregion
 
 
+def test_pass_through_serdes():
+    serdes = PassThroughSerDes()
+
+    data = '"name": "test", "value": 123'
+    serialized = serialize(serdes, data, "test-op", "test-arn")
+    assert isinstance(serialized, str)
+    assert serialized == '"name": "test", "value": 123'
+    # Dict uses envelope format, so roundtrip through deserialize
+    deserialized = deserialize(serdes, serialized, "test-op", "test-arn")
+    assert deserialized == data
+
+
 # region EnvelopeSerDes Performance and Edge Cases
 def test_envelope_large_data_structure():
     """Test with reasonably large data."""