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,17 +21,37 @@
     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,
 )
 from aws_durable_execution_sdk_python_testing.token import CheckpointToken
 
 
+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"
+
+
 class Execution:
     """Execution state."""
 
@@ -52,12 +73,39 @@ 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
 
+    @property
+    def status(self) -> str:
+        """Get execution status string (mimics backend ExecutionStatusConverter)."""
+        if not self.is_complete:
+            return ExecutionStatus.RUNNING.value
+
+        if not self.close_status:
+            msg: str = "close_status cannot be None"
+            raise IllegalStateException(msg)
+
+        # Convert CloseStatus to ExecutionStatus (like backend ExecutionStatusConverter)
+        match self.close_status:
+            case CloseStatus.COMPLETED:
+                return ExecutionStatus.SUCCEEDED.value
+            case CloseStatus.FAILED:
+                return ExecutionStatus.FAILED.value
+            case CloseStatus.TERMINATED:
+                return ExecutionStatus.STOPPED.value
+            case CloseStatus.TIMED_OUT:
+                return ExecutionStatus.TIMED_OUT.value
+            case _:
+                msg: str = f"Unexpected close status: {self.close_status}"
+                raise InvalidParameterValueException(msg)
+
     @staticmethod
     def new(input: StartDurableExecutionInput) -> Execution:  # noqa: A002
         # make a nicer arn
@@ -79,6 +127,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
@@ -112,6 +161,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
 
@@ -184,16 +237,36 @@ 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
 
     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
+
+    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
+
+    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
 
     def find_operation(self, operation_id: str) -> tuple[int, Operation]:
         """Find operation by ID, return index and operation."""