feat: execution state pagination and token validation

- Add paging logic in checkpoint processor with next_marker support
- Implement checkpoint token validation
- Add token expiration checking with error responses
- Handle missing token cases with context-appropriate validation
- Add pagination metadata to responses with configurable max_items
- Add test coverage for all validation scenarios

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py

@@ -88,14 +88,35 @@ def process_checkpoint(
     def get_execution_state(
         self,
         checkpoint_token: str,
-        next_marker: str,  # noqa: ARG002
-        max_items: int = 1000,  # noqa: ARG002
+        next_marker: str | None = None,
+        max_items: int = 1000,
     ) -> StateOutput:
-        """Get current execution state."""
+        """Get current execution state with batched checkpoint token validation and pagination."""
+        if not checkpoint_token:
+            msg: str = "Checkpoint token is required"
+            raise InvalidParameterValueException(msg)
+
         token: CheckpointToken = CheckpointToken.from_str(checkpoint_token)
         execution: Execution = self._store.load(token.execution_arn)
+        execution.validate_checkpoint_token(checkpoint_token)
+
+        # Get all operations
+        all_operations = execution.get_navigable_operations()
+
+        # Apply pagination
+        start_index = 0
+        if next_marker:
+            try:
+                start_index = int(next_marker)
+            except ValueError:
+                start_index = 0
+
+        end_index = start_index + max_items
+        paginated_operations = all_operations[start_index:end_index]
+
+        # Determine next marker
+        next_marker_result = str(end_index) if end_index < len(all_operations) else None
 
-        # TODO: paging when size or max
         return StateOutput(
-            operations=execution.get_navigable_operations(), next_marker=None
+            operations=paginated_operations, next_marker=next_marker_result
         )

src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py

@@ -38,12 +38,8 @@ def process(
                 )
             case _:
                 # intentional. actual service will fail any EXECUTION update that is not SUCCEED.
-                error = (
-                    update.error
-                    if update.error
-                    else ErrorObject.from_message(
-                        "There is no error details but EXECUTION checkpoint action is not SUCCEED."
-                    )
+                error = update.error or ErrorObject.from_message(
+                    "There is no error details but EXECUTION checkpoint action is not SUCCEED."
                 )
                 notifier.notify_failed(execution_arn=execution_arn, error=error)
         # TODO: Svc doesn't actually create checkpoint for EXECUTION. might have to for localrunner though.

src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py

@@ -55,7 +55,7 @@ def __init__(
         self,
         processors: MutableMapping[OperationType, OperationProcessor] | None = None,
     ):
-        self.processors = processors if processors else self._DEFAULT_PROCESSORS
+        self.processors = processors or self._DEFAULT_PROCESSORS
 
     def process_updates(
         self,

src/aws_durable_execution_sdk_python_testing/execution.py

@@ -159,6 +159,35 @@ def get_navigable_operations(self) -> list[Operation]:
         """Get list of operations, but exclude child operations where the parent has already completed."""
         return self.operations
 
+    def validate_checkpoint_token(
+        self,
+        checkpoint_token: str | None,
+        checkpoint_required: bool = True,  # noqa: FBT001, FBT002
+        checkpoint_required_msg: str | None = None,
+    ) -> None:
+        """Validate checkpoint token against this execution."""
+        if not checkpoint_token and checkpoint_required:
+            msg: str = checkpoint_required_msg or "Checkpoint token is required"
+            raise InvalidParameterValueException(msg)
+
+        if not checkpoint_token:
+            return  # No token provided and not required
+
+        token: CheckpointToken = CheckpointToken.from_str(checkpoint_token)
+        if token.execution_arn != self.durable_execution_arn:
+            msg = "Checkpoint token does not match execution ARN"
+            raise InvalidParameterValueException(msg)
+
+        if self.is_complete or token.token_sequence > self.token_sequence:
+            msg = "Invalid or expired checkpoint token"
+            raise InvalidParameterValueException(msg)
+
+        # Check if token has been used
+        token_str: str = token.to_str()
+        if token_str not in self.used_tokens:
+            msg = f"Invalid checkpoint token: {token_str}"
+            raise InvalidParameterValueException(msg)
+
     def get_assertable_operations(self) -> list[Operation]:
         """Get list of operations, but exclude the EXECUTION operations"""
         # TODO: this excludes EXECUTION at start, but can there be an EXECUTION at the end if there was a checkpoint with large payload?

src/aws_durable_execution_sdk_python_testing/executor.py

@@ -46,6 +46,8 @@
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
+    from aws_durable_execution_sdk_python.lambda_service import Operation
+
     from aws_durable_execution_sdk_python_testing.invoker import Invoker
     from aws_durable_execution_sdk_python_testing.scheduler import Event, Scheduler
     from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore
@@ -357,32 +359,33 @@ def get_execution_state(
             ResourceNotFoundException: If execution does not exist
             InvalidParameterValueException: If checkpoint token is invalid
         """
-        execution = self.get_execution(execution_arn)
-
-        # TODO: Validate checkpoint token if provided
-        if checkpoint_token and checkpoint_token not in execution.used_tokens:
-            msg: str = f"Invalid checkpoint token: {checkpoint_token}"
-            raise InvalidParameterValueException(msg)
+        execution: Execution = self.get_execution(execution_arn)
+        checkpoint_required_msg: str = (
+            "Checkpoint token is required for paginated requests on active executions"
+        )
+        checkpoint_required: bool = not execution.is_complete and marker is not None
+        execution.validate_checkpoint_token(
+            checkpoint_token, checkpoint_required, checkpoint_required_msg
+        )
 
         # Get operations (excluding the initial EXECUTION operation for state)
-        operations = execution.get_assertable_operations()
+        operations: list[Operation] = execution.get_assertable_operations()
 
         # Apply pagination
         if max_items is None:
             max_items = 100
 
-        # Simple pagination - in real implementation would need proper marker handling
-        start_index = 0
+        start_index: int = 0
         if marker:
             try:
                 start_index = int(marker)
             except ValueError:
                 start_index = 0
 
-        end_index = start_index + max_items
-        paginated_operations = operations[start_index:end_index]
+        end_index: int = start_index + max_items
+        paginated_operations: list[Operation] = operations[start_index:end_index]
 
-        next_marker = None
+        next_marker: str | None = None
         if end_index < len(operations):
             next_marker = str(end_index)
 
@@ -467,11 +470,11 @@ def checkpoint_execution(
             InvalidParameterValueException: If checkpoint token is invalid
         """
         execution = self.get_execution(execution_arn)
-
-        # Validate checkpoint token
-        if checkpoint_token not in execution.used_tokens:
-            msg: str = f"Invalid checkpoint token: {checkpoint_token}"
-            raise InvalidParameterValueException(msg)
+        execution.validate_checkpoint_token(
+            checkpoint_token,
+            checkpoint_required=True,
+            checkpoint_required_msg="Checkpoint token is required for checkpoint operations",
+        )
 
         # TODO: Process operation updates using the checkpoint processor
         # This would integrate with the existing checkpoint processing pipeline