feat: complete sqlite store and function handler implementation

- Add SQLiteExecutionStore with database persistence and indexing
- Implement query system with pagination support
- Add BaseExecutionStore with shared query processing logic
- Update Executor to use new query system for efficient operations
- Complete ListDurableExecutionsByFunctionHandler with proper filtering
- Add function name validation and error handling
- Add comprehensive test coverage for all implementations
- Support concurrent access patterns with proper database handling

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py

@@ -45,6 +45,8 @@ def process(
                         "There is no error details but EXECUTION checkpoint action is not SUCCEED."
                     )
                 )
+                # All EXECUTION failures go through normal fail path
+                # Timeout/Stop status is set by executor based on the operation that caused it
                 notifier.notify_failed(execution_arn=execution_arn, error=error)
         # TODO: Svc doesn't actually create checkpoint for EXECUTION. might have to for localrunner though.
         return None

src/aws_durable_execution_sdk_python_testing/execution.py

@@ -3,6 +3,7 @@
 import json
 from dataclasses import replace
 from datetime import UTC, datetime
+from enum import Enum
 from threading import Lock
 from typing import Any
 from uuid import uuid4
@@ -20,11 +21,12 @@
     OperationUpdate,
 )
 
-# Import AWS exceptions
 from aws_durable_execution_sdk_python_testing.exceptions import (
     IllegalStateException,
     InvalidParameterValueException,
 )
+
+# Import AWS exceptions
 from aws_durable_execution_sdk_python_testing.model import (
     StartDurableExecutionInput,
 )
@@ -34,6 +36,41 @@
 )
 
 
+class CloseStatus(Enum):
+    """Close status for completed executions (mimics backend SWF CloseStatus)."""
+
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    TERMINATED = "TERMINATED"
+    TIMED_OUT = "TIMED_OUT"
+
+
+class ExecutionStatus(Enum):
+    """Execution status for API responses (mimics backend ExecutionStatus)."""
+
+    RUNNING = "RUNNING"
+    SUCCEEDED = "SUCCEEDED"
+    FAILED = "FAILED"
+    STOPPED = "STOPPED"
+    TIMED_OUT = "TIMED_OUT"
+
+    @classmethod
+    def from_close_status(cls, close_status: CloseStatus) -> ExecutionStatus:
+        """Convert CloseStatus to ExecutionStatus."""
+        match close_status:
+            case CloseStatus.COMPLETED:
+                return cls.SUCCEEDED
+            case CloseStatus.FAILED:
+                return cls.FAILED
+            case CloseStatus.TERMINATED:
+                return cls.STOPPED
+            case CloseStatus.TIMED_OUT:
+                return cls.TIMED_OUT
+            case _:
+                error_msg: str = f"Unexpected close status: {close_status}"
+                raise InvalidParameterValueException(error_msg)
+
+
 class Execution:
     """Execution state."""
 
@@ -55,12 +92,27 @@ def __init__(
         self.is_complete: bool = False
         self.result: DurableExecutionInvocationOutput | None = None
         self.consecutive_failed_invocation_attempts: int = 0
+        self.close_status: CloseStatus | None = (
+            None  # Track close status like backend SWF
+        )
 
     @property
     def token_sequence(self) -> int:
         """Get current token sequence value."""
         return self._token_sequence
 
+    def current_status(self) -> ExecutionStatus:
+        """Get execution status."""
+        if not self.is_complete:
+            return ExecutionStatus.RUNNING
+
+        if not self.close_status:
+            msg: str = "close_status must be set when execution is complete"
+            raise IllegalStateException(msg)
+
+        # Convert CloseStatus to ExecutionStatus
+        return ExecutionStatus.from_close_status(self.close_status)
+
     @staticmethod
     def new(input: StartDurableExecutionInput) -> Execution:  # noqa: A002
         # make a nicer arn
@@ -82,6 +134,7 @@ def to_dict(self) -> dict[str, Any]:
             "IsComplete": self.is_complete,
             "Result": self.result.to_dict() if self.result else None,
             "ConsecutiveFailedInvocationAttempts": self.consecutive_failed_invocation_attempts,
+            "CloseStatus": self.close_status.value if self.close_status else None,
         }
 
     @classmethod
@@ -115,6 +168,10 @@ def from_dict(cls, data: dict[str, Any]) -> Execution:
         execution.consecutive_failed_invocation_attempts = data[
             "ConsecutiveFailedInvocationAttempts"
         ]
+        close_status_str = data.get("CloseStatus")
+        execution.close_status = (
+            CloseStatus(close_status_str) if close_status_str else None
+        )
 
         return execution
 
@@ -187,16 +244,40 @@ def has_pending_operations(self, execution: Execution) -> bool:
         return False
 
     def complete_success(self, result: str | None) -> None:
+        """Complete execution successfully (DecisionType.COMPLETE_WORKFLOW_EXECUTION)."""
         self.result = DurableExecutionInvocationOutput(
             status=InvocationStatus.SUCCEEDED, result=result
         )
         self.is_complete = True
+        self.close_status = CloseStatus.COMPLETED
+        self._end_execution(OperationStatus.SUCCEEDED)
 
     def complete_fail(self, error: ErrorObject) -> None:
+        """Complete execution with failure (DecisionType.FAIL_WORKFLOW_EXECUTION)."""
+        self.result = DurableExecutionInvocationOutput(
+            status=InvocationStatus.FAILED, error=error
+        )
+        self.is_complete = True
+        self.close_status = CloseStatus.FAILED
+        self._end_execution(OperationStatus.FAILED)
+
+    def complete_timeout(self, error: ErrorObject) -> None:
+        """Complete execution with timeout (SWF workflow timeout)."""
         self.result = DurableExecutionInvocationOutput(
             status=InvocationStatus.FAILED, error=error
         )
         self.is_complete = True
+        self.close_status = CloseStatus.TIMED_OUT
+        self._end_execution(OperationStatus.TIMED_OUT)
+
+    def complete_stopped(self, error: ErrorObject) -> None:
+        """Complete execution as terminated (TerminateWorkflowExecutionV2Request)."""
+        self.result = DurableExecutionInvocationOutput(
+            status=InvocationStatus.FAILED, error=error
+        )
+        self.is_complete = True
+        self.close_status = CloseStatus.TERMINATED
+        self._end_execution(OperationStatus.STOPPED)
 
     def find_operation(self, operation_id: str) -> tuple[int, Operation]:
         """Find operation by ID, return index and operation."""
@@ -327,3 +408,14 @@ def complete_callback_failure(
                 callback_details=updated_callback_details,
             )
             return self.operations[index]
+
+    def _end_execution(self, status: OperationStatus) -> None:
+        """Set the end_timestamp on the main EXECUTION operation when execution completes."""
+        execution_op: Operation = self.get_operation_execution_started()
+        if execution_op.operation_type == OperationType.EXECUTION:
+            with self._state_lock:
+                self.operations[0] = replace(
+                    execution_op,
+                    status=status,
+                    end_timestamp=datetime.now(UTC),
+                )