refactor: replace dataclasses.replace with explicit factory methods

- Add Operation.from_existing() classmethod
- Add StepDetails.from_existing() and CallbackDetails.from_existing() factory methods
- Replace all dataclasses.replace() calls with explicit constructors
- Add test coverage for new factory methods

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python/lambda_service.py

@@ -187,6 +187,20 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> StepDetails:
             error=ErrorObject.from_dict(error_raw) if error_raw else None,
         )
 
+    @classmethod
+    def from_existing(
+        cls,
+        step_details: StepDetails,
+        next_attempt_timestamp: datetime.datetime | None = None,
+    ) -> StepDetails:
+        """Create new StepDetails from existing with cleared next_attempt_timestamp."""
+        return cls(
+            attempt=step_details.attempt,
+            next_attempt_timestamp=next_attempt_timestamp,
+            result=step_details.result,
+            error=step_details.error,
+        )
+
 
 @dataclass(frozen=True)
 class WaitDetails:
@@ -212,6 +226,20 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> CallbackDetails:
             error=ErrorObject.from_dict(error_raw) if error_raw else None,
         )
 
+    @classmethod
+    def from_existing(
+        cls,
+        callback_details: CallbackDetails,
+        result: str | None = None,
+        error: ErrorObject | None = None,
+    ) -> CallbackDetails:
+        """Create new CallbackDetails from existing with updated result or error."""
+        return cls(
+            callback_id=callback_details.callback_id,
+            result=callback_details.result if result is None else result,
+            error=callback_details.error if error is None else error,
+        )
+
 
 @dataclass(frozen=True)
 class ChainedInvokeDetails:
@@ -708,6 +736,57 @@ class Operation:
     callback_details: CallbackDetails | None = None
     chained_invoke_details: ChainedInvokeDetails | None = None
 
+    @classmethod
+    def from_existing(
+        cls,
+        operation: Operation,
+        status: OperationStatus | None = None,
+        name: str | None = None,
+        parent_id: str | None = None,
+        sub_type: OperationSubType | None = None,
+        start_timestamp: datetime.datetime | None = None,
+        end_timestamp: datetime.datetime | None = None,
+        execution_details: ExecutionDetails | None = None,
+        context_details: ContextDetails | None = None,
+        step_details: StepDetails | None = None,
+        wait_details: WaitDetails | None = None,
+        callback_details: CallbackDetails | None = None,
+        chained_invoke_details: ChainedInvokeDetails | None = None,
+    ) -> Operation:
+        """Create a new Operation by copying an existing one with selective field updates."""
+        return cls(
+            operation_id=operation.operation_id,
+            operation_type=operation.operation_type,
+            status=operation.status if status is None else status,
+            parent_id=operation.parent_id if parent_id is None else parent_id,
+            name=operation.name if name is None else name,
+            start_timestamp=operation.start_timestamp
+            if start_timestamp is None
+            else start_timestamp,
+            end_timestamp=operation.end_timestamp
+            if end_timestamp is None
+            else end_timestamp,
+            sub_type=operation.sub_type if sub_type is None else sub_type,
+            execution_details=operation.execution_details
+            if execution_details is None
+            else execution_details,
+            context_details=operation.context_details
+            if context_details is None
+            else context_details,
+            step_details=operation.step_details
+            if step_details is None
+            else step_details,
+            wait_details=operation.wait_details
+            if wait_details is None
+            else wait_details,
+            callback_details=operation.callback_details
+            if callback_details is None
+            else callback_details,
+            chained_invoke_details=operation.chained_invoke_details
+            if chained_invoke_details is None
+            else chained_invoke_details,
+        )
+
     @classmethod
     def from_dict(cls, data: MutableMapping[str, Any]) -> Operation:
         """Create an Operation instance from a dictionary with the original Smithy model field names.

tests/lambda_service_test.py

@@ -277,6 +277,23 @@ def test_step_details_minimal():
     assert details.error is None
 
 
+def test_step_details_from_existing():
+    """Test StepDetails.from_existing factory method."""
+    original = StepDetails(
+        attempt=2,
+        next_attempt_timestamp=datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC),
+        result="original_result",
+        error=ErrorObject.from_message("original_error"),
+    )
+
+    # Test clearing next_attempt_timestamp
+    updated = StepDetails.from_existing(original, next_attempt_timestamp=None)
+    assert updated.attempt == 2
+    assert updated.next_attempt_timestamp is None
+    assert updated.result == "original_result"
+    assert updated.error.message == "original_error"
+
+
 def test_wait_details_from_dict():
     """Test WaitDetails.from_dict method."""
     timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC)
@@ -330,6 +347,36 @@ def test_callback_details_minimal():
     assert details.error is None
 
 
+def test_callback_details_from_existing_with_result():
+    """Test CallbackDetails.from_existing factory method with result update."""
+    original = CallbackDetails(
+        callback_id="cb123",
+        result="original_result",
+        error=ErrorObject.from_message("original_error"),
+    )
+
+    # Test updating result
+    updated = CallbackDetails.from_existing(original, result="new_result")
+    assert updated.callback_id == "cb123"
+    assert updated.result == "new_result"
+    assert updated.error.message == "original_error"
+
+
+def test_callback_details_from_existing_with_error():
+    """Test CallbackDetails.from_existing factory method with error update."""
+    original = CallbackDetails(
+        callback_id="cb123",
+        result="original_result",
+        error=None,
+    )
+
+    new_error = ErrorObject.from_message("new_error")
+    updated = CallbackDetails.from_existing(original, error=new_error)
+    assert updated.callback_id == "cb123"
+    assert updated.result == "original_result"
+    assert updated.error.message == "new_error"
+
+
 def test_invoke_details_from_dict():
     """Test ChainedInvokeDetails.from_dict method."""
     error_data = {"ErrorMessage": "Invoke error"}
@@ -1119,6 +1166,27 @@ def test_operation_from_dict_no_options():
     assert operation.operation_id == "test-id"
 
 
+def test_operation_from_existing_with_callback_details():
+    """Test Operation.from_existing factory method with callback_details update."""
+    original = Operation(
+        operation_id="op1",
+        operation_type=OperationType.CALLBACK,
+        status=OperationStatus.STARTED,
+        callback_details=CallbackDetails(callback_id="cb123", result=None),
+    )
+
+    new_callback_details = CallbackDetails(callback_id="cb123", result="success_result")
+    updated = Operation.from_existing(
+        original,
+        status=OperationStatus.SUCCEEDED,
+        callback_details=new_callback_details,
+    )
+
+    assert updated.status == OperationStatus.SUCCEEDED
+    assert updated.callback_details.callback_id == "cb123"
+    assert updated.callback_details.result == "success_result"
+
+
 def test_operation_from_dict_individual_options():
     """Test Operation.from_dict with each option type individually."""
     # Test with just ContextOptions
@@ -1510,6 +1578,60 @@ def test_operation_to_dict_all_optional_fields():
     assert result["SubType"] == "Step"
 
 
+def test_operation_from_existing():
+    """Test Operation.from_existing factory method."""
+    original = Operation(
+        operation_id="op1",
+        operation_type=OperationType.STEP,
+        status=OperationStatus.STARTED,
+        parent_id="parent1",
+        name="original_step",
+        start_timestamp=datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC),
+        step_details=StepDetails(attempt=1, result="original_result"),
+    )
+
+    # Test updating status and end_timestamp
+    updated = Operation.from_existing(
+        original,
+        status=OperationStatus.SUCCEEDED,
+        end_timestamp=datetime.datetime(2023, 1, 2, tzinfo=datetime.UTC),
+    )
+
+    assert updated.operation_id == "op1"
+    assert updated.operation_type == OperationType.STEP
+    assert updated.status == OperationStatus.SUCCEEDED
+    assert updated.parent_id == "parent1"
+    assert updated.name == "original_step"
+    assert updated.start_timestamp == datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC)
+    assert updated.end_timestamp == datetime.datetime(2023, 1, 2, tzinfo=datetime.UTC)
+    assert updated.step_details.attempt == 1
+    assert updated.step_details.result == "original_result"
+
+
+def test_operation_from_existing_with_step_details():
+    """Test Operation.from_existing factory method with step_details update."""
+    original = Operation(
+        operation_id="op1",
+        operation_type=OperationType.STEP,
+        status=OperationStatus.PENDING,
+        step_details=StepDetails(
+            attempt=1,
+            next_attempt_timestamp=datetime.datetime(2023, 1, 1, tzinfo=datetime.UTC),
+        ),
+    )
+
+    new_step_details = StepDetails(attempt=1, next_attempt_timestamp=None)
+    updated = Operation.from_existing(
+        original,
+        status=OperationStatus.READY,
+        step_details=new_step_details,
+    )
+
+    assert updated.status == OperationStatus.READY
+    assert updated.step_details.attempt == 1
+    assert updated.step_details.next_attempt_timestamp is None
+
+
 # =============================================================================
 # Tests for Checkpoint Classes
 # =============================================================================