feat: implement callback token generation and processing

- Add CallbackToken generation in callback processor with observer integration
- Implement SendCallbackSuccess, SendCallbackFailure, and SendCallbackHeartbeat
- Add callback operation lookup and completion methods to execution
- Ensure unique token generation across executions

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py

@@ -94,12 +94,12 @@ def _create_step_details(
         return None
 
     def _create_callback_details(
-        self, update: OperationUpdate
+        self, update: OperationUpdate, callback_id: str | None = None
     ) -> CallbackDetails | None:
         """Create CallbackDetails from OperationUpdate."""
         return (
             CallbackDetails(
-                callback_id="placeholder", result=update.payload, error=update.error
+                callback_id=callback_id or "", result=update.payload, error=update.error
             )
             if update.operation_type == OperationType.CALLBACK
             else None
@@ -137,6 +137,7 @@ def _translate_update_to_operation(
         update: OperationUpdate,
         current_operation: Operation | None,
         status: OperationStatus,
+        callback_id: str | None = None,
     ) -> Operation:
         """Transform OperationUpdate to Operation, always creating new Operation."""
         start_time = (
@@ -149,7 +150,7 @@ def _translate_update_to_operation(
         execution_details = self._create_execution_details(update)
         context_details = self._create_context_details(update)
         step_details = self._create_step_details(update, current_operation)
-        callback_details = self._create_callback_details(update)
+        callback_details = self._create_callback_details(update, callback_id)
         invoke_details = self._create_invoke_details(update)
         wait_details = self._create_wait_details(update, current_operation)
 

src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py

@@ -2,13 +2,15 @@
 
 from __future__ import annotations
 
+from dataclasses import replace
 from typing import TYPE_CHECKING
 
 from aws_durable_execution_sdk_python.lambda_service import (
     Operation,
     OperationAction,
     OperationStatus,
     OperationUpdate,
+    CallbackDetails,
 )
 
 from aws_durable_execution_sdk_python_testing.checkpoint.processors.base import (
@@ -17,6 +19,7 @@
 from aws_durable_execution_sdk_python_testing.exceptions import (
     InvalidParameterValueException,
 )
+from aws_durable_execution_sdk_python_testing.token import CallbackToken
 
 
 if TYPE_CHECKING:
@@ -36,14 +39,26 @@ def process(
         """Process CALLBACK operation update with scheduler integration for activities."""
         match update.action:
             case OperationAction.START:
-                # TODO: create CallbackToken (see token module). Add Observer/Notifier for on_callback_created possibly,
-                # but token might well have enough so don't need to maintain token list on execution itself
-                return self._translate_update_to_operation(
+                callback_token: CallbackToken = CallbackToken(
+                    execution_arn=execution_arn,
+                    operation_id=update.operation_id,
+                )
+                callback_id: str = callback_token.to_str()
+
+                notifier.notify_callback_created(
+                    execution_arn=execution_arn,
+                    operation_id=update.operation_id,
+                    callback_id=callback_id,
+                )
+
+                operation: Operation = self._translate_update_to_operation(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.STARTED,
+                    callback_id=callback_id,
                 )
+
+                return operation
             case _:
                 msg: str = "Invalid action for CALLBACK operation."
-
                 raise InvalidParameterValueException(msg)

src/aws_durable_execution_sdk_python_testing/checkpoint/validators/operations/callback.py

@@ -17,7 +17,6 @@
 VALID_ACTIONS_FOR_CALLBACK = frozenset(
     [
         OperationAction.START,
-        OperationAction.CANCEL,
     ]
 )
 
@@ -41,14 +40,6 @@ def validate(current_state: Operation | None, update: OperationUpdate) -> None:
                         "Cannot start a CALLBACK that already exist."
                     )
                     raise InvalidParameterValueException(msg_callback_exists)
-            case OperationAction.CANCEL:
-                if (
-                    current_state is None
-                    or current_state.status
-                    not in CallbackOperationValidator._ALLOWED_STATUS_TO_CANCEL
-                ):
-                    msg_callback_cancel: str = "Cannot cancel a CALLBACK that does not exist or has already completed."
-                    raise InvalidParameterValueException(msg_callback_cancel)
             case _:
-                msg_callback_invalid: str = "Invalid CALLBACK action."
-                raise InvalidParameterValueException(msg_callback_invalid)
+                msg: str = "Invalid action for CALLBACK operation."
+                raise InvalidParameterValueException(msg)

src/aws_durable_execution_sdk_python_testing/execution.py

@@ -28,7 +28,10 @@
 from aws_durable_execution_sdk_python_testing.model import (
     StartDurableExecutionInput,
 )
-from aws_durable_execution_sdk_python_testing.token import CheckpointToken
+from aws_durable_execution_sdk_python_testing.token import (
+    CheckpointToken,
+    CallbackToken,
+)
 
 
 class Execution:
@@ -203,6 +206,18 @@ def find_operation(self, operation_id: str) -> tuple[int, Operation]:
         msg: str = f"Attempting to update state of an Operation [{operation_id}] that doesn't exist"
         raise IllegalStateException(msg)
 
+    def find_callback_operation(self, callback_id: str) -> tuple[int, Operation]:
+        """Find callback operation by callback_id, return index and operation."""
+        for i, operation in enumerate(self.operations):
+            if (
+                operation.operation_type == OperationType.CALLBACK
+                and operation.callback_details
+                and operation.callback_details.callback_id == callback_id
+            ):
+                return i, operation
+        msg: str = f"Callback operation with callback_id [{callback_id}] not found"
+        raise IllegalStateException(msg)
+
     def complete_wait(self, operation_id: str) -> Operation:
         """Complete WAIT operation when timer fires."""
         index, operation = self.find_operation(operation_id)
@@ -260,3 +275,56 @@ def complete_retry(self, operation_id: str) -> Operation:
             # Assign
             self.operations[index] = updated_operation
             return updated_operation
+
+    def complete_callback_success(
+        self, callback_id: str, result: bytes | None = None
+    ) -> Operation:
+        """Complete CALLBACK operation with success."""
+        index, operation = self.find_callback_operation(callback_id)
+
+        if operation.status != OperationStatus.STARTED:
+            msg: str = f"Callback operation [{callback_id}] is not in STARTED state"
+            raise IllegalStateException(msg)
+
+        with self._state_lock:
+            self._token_sequence += 1
+            updated_callback_details = None
+            if operation.callback_details:
+                updated_callback_details = replace(
+                    operation.callback_details,
+                    result=result.decode() if result else None,
+                )
+
+            self.operations[index] = replace(
+                operation,
+                status=OperationStatus.SUCCEEDED,
+                end_timestamp=datetime.now(UTC),
+                callback_details=updated_callback_details,
+            )
+            return self.operations[index]
+
+    def complete_callback_failure(
+        self, callback_id: str, error: ErrorObject
+    ) -> Operation:
+        """Complete CALLBACK operation with failure."""
+        index, operation = self.find_callback_operation(callback_id)
+
+        if operation.status != OperationStatus.STARTED:
+            msg: str = f"Callback operation [{callback_id}] is not in STARTED state"
+            raise IllegalStateException(msg)
+
+        with self._state_lock:
+            self._token_sequence += 1
+            updated_callback_details = None
+            if operation.callback_details:
+                updated_callback_details = replace(
+                    operation.callback_details, error=error
+                )
+
+            self.operations[index] = replace(
+                operation,
+                status=OperationStatus.FAILED,
+                end_timestamp=datetime.now(UTC),
+                callback_details=updated_callback_details,
+            )
+            return self.operations[index]