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,38 @@ 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 (like backend ExecutionStatusConverter)
+        match self.close_status:
+            case CloseStatus.COMPLETED:
+                return ExecutionStatus.SUCCEEDED
+            case CloseStatus.FAILED:
+                return ExecutionStatus.FAILED
+            case CloseStatus.TERMINATED:
+                return ExecutionStatus.STOPPED
+            case CloseStatus.TIMED_OUT:
+                return ExecutionStatus.TIMED_OUT
+            case _:
+                error_msg: str = f"Unexpected close status: {self.close_status}"
+                raise InvalidParameterValueException(error_msg)
+
     @staticmethod
     def new(input: StartDurableExecutionInput) -> Execution:  # noqa: A002
         # make a nicer arn
@@ -79,6 +126,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 +160,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 +236,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."""

src/aws_durable_execution_sdk_python_testing/executor.py

@@ -142,18 +142,7 @@ def get_execution_details(self, execution_arn: str) -> GetDurableExecutionRespon
 
         # Extract execution details from the first operation (EXECUTION type)
         execution_op = execution.get_operation_execution_started()
-
-        # Determine status based on execution state
-        if execution.is_complete:
-            if (
-                execution.result
-                and execution.result.status == InvocationStatus.SUCCEEDED
-            ):
-                status = "SUCCEEDED"
-            else:
-                status = "FAILED"
-        else:
-            status = "RUNNING"
+        status = execution.current_status().value
 
         # Extract result and error from execution result
         result = None
@@ -189,8 +178,8 @@ def list_executions(
         function_version: str | None = None,  # noqa: ARG002
         execution_name: str | None = None,
         status_filter: str | None = None,
-        time_after: str | None = None,  # noqa: ARG002
-        time_before: str | None = None,  # noqa: ARG002
+        started_after: str | None = None,
+        started_before: str | None = None,
         marker: str | None = None,
         max_items: int | None = None,
         reverse_order: bool = False,  # noqa: FBT001, FBT002
@@ -202,86 +191,43 @@ def list_executions(
             function_version: Filter by function version
             execution_name: Filter by execution name
             status_filter: Filter by status (RUNNING, SUCCEEDED, FAILED)
-            time_after: Filter executions started after this time
-            time_before: Filter executions started before this time
+            started_after: Filter executions started after this time
+            started_before: Filter executions started before this time
             marker: Pagination marker
             max_items: Maximum items to return (default 50)
             reverse_order: Return results in reverse chronological order
 
         Returns:
             ListDurableExecutionsResponse: List of executions with pagination
         """
-        # Get all executions from store
-        all_executions = self._store.list_all()
-
-        # Apply filters
-        filtered_executions = []
-        for execution in all_executions:
-            # Filter by function name
-            if function_name and execution.start_input.function_name != function_name:
-                continue
-
-            # Filter by execution name
-            if (
-                execution_name
-                and execution.start_input.execution_name != execution_name
-            ):
-                continue
-
-            # Determine execution status
-            execution_status = "RUNNING"
-            if execution.is_complete:
-                if (
-                    execution.result
-                    and execution.result.status == InvocationStatus.SUCCEEDED
-                ):
-                    execution_status = "SUCCEEDED"
-                else:
-                    execution_status = "FAILED"
-
-            # Filter by status
-            if status_filter and execution_status != status_filter:
-                continue
-
-            # Convert to ExecutionSummary
-            execution_op = execution.get_operation_execution_started()
-            execution_summary = ExecutionSummary(
-                durable_execution_arn=execution.durable_execution_arn,
-                durable_execution_name=execution.start_input.execution_name,
-                function_arn=f"arn:aws:lambda:us-east-1:123456789012:function:{execution.start_input.function_name}",
-                status=execution_status,
-                start_timestamp=execution_op.start_timestamp
-                if execution_op.start_timestamp
-                else datetime.now(UTC),
-                end_timestamp=execution_op.end_timestamp
-                if execution_op.end_timestamp
-                else None,
-            )
-            filtered_executions.append(execution_summary)
-
-        # Sort by start date
-        filtered_executions.sort(key=lambda e: e.start_timestamp, reverse=reverse_order)
-
-        # Apply pagination
-        if max_items is None:
-            max_items = 50
-
-        start_index = 0
+        # Convert marker to offset
+        offset: int = 0
         if marker:
             try:
-                start_index = int(marker)
+                offset = int(marker)
             except ValueError:
-                start_index = 0
+                offset = 0
 
-        end_index = start_index + max_items
-        paginated_executions = filtered_executions[start_index:end_index]
+        # Query store directly with parameters
+        executions, next_marker = self._store.query(
+            function_name=function_name,
+            execution_name=execution_name,
+            status_filter=status_filter,
+            started_after=started_after,
+            started_before=started_before,
+            limit=max_items or 50,
+            offset=offset,
+            reverse_order=reverse_order,
+        )
 
-        next_marker = None
-        if end_index < len(filtered_executions):
-            next_marker = str(end_index)
+        # Convert to ExecutionSummary objects
+        execution_summaries: list[ExecutionSummary] = [
+            ExecutionSummary.from_execution(execution, execution.current_status().value)
+            for execution in executions
+        ]
 
         return ListDurableExecutionsResponse(
-            durable_executions=paginated_executions, next_marker=next_marker
+            durable_executions=execution_summaries, next_marker=next_marker
         )
 
     def list_executions_by_function(
@@ -290,8 +236,8 @@ def list_executions_by_function(
         qualifier: str | None = None,  # noqa: ARG002
         execution_name: str | None = None,
         status_filter: str | None = None,
-        time_after: str | None = None,
-        time_before: str | None = None,
+        started_after: str | None = None,
+        started_before: str | None = None,
         marker: str | None = None,
         max_items: int | None = None,
         reverse_order: bool = False,  # noqa: FBT001, FBT002
@@ -303,8 +249,8 @@ def list_executions_by_function(
             qualifier: Function qualifier/version
             execution_name: Filter by execution name
             status_filter: Filter by status (RUNNING, SUCCEEDED, FAILED)
-            time_after: Filter executions started after this time
-            time_before: Filter executions started before this time
+            started_after: Filter executions started after this time
+            started_before: Filter executions started before this time
             marker: Pagination marker
             max_items: Maximum items to return (default 50)
             reverse_order: Return results in reverse chronological order
@@ -317,8 +263,8 @@ def list_executions_by_function(
             function_name=function_name,
             execution_name=execution_name,
             status_filter=status_filter,
-            time_after=time_after,
-            time_before=time_before,
+            started_after=started_after,
+            started_before=started_before,
             marker=marker,
             max_items=max_items,
             reverse_order=reverse_order,
@@ -357,8 +303,11 @@ def stop_execution(
             "Execution stopped by user request"
         )
 
-        # Stop the execution
-        self.fail_execution(execution_arn, stop_error)
+        # Stop sets TERMINATED close status (different from fail)
+        logger.exception("[%s] Stopping execution.", execution_arn)
+        execution.complete_stopped(error=stop_error)  # Sets CloseStatus.TERMINATED
+        self._store.update(execution)
+        self._complete_events(execution_arn=execution_arn)
 
         return StopDurableExecutionResponse(stop_timestamp=datetime.now(UTC))
 
@@ -808,27 +757,24 @@ def wait_until_complete(
         raise ResourceNotFoundException(msg)
 
     def complete_execution(self, execution_arn: str, result: str | None = None) -> None:
-        """Complete execution successfully."""
+        """Complete execution successfully (COMPLETE_WORKFLOW_EXECUTION decision)."""
         logger.debug("[%s] Completing execution with result: %s", execution_arn, result)
         execution: Execution = self._store.load(execution_arn=execution_arn)
-        execution.complete_success(result=result)
+        execution.complete_success(result=result)  # Sets CloseStatus.COMPLETED
         self._store.update(execution)
         if execution.result is None:
             msg: str = "Execution result is required"
-
             raise IllegalStateException(msg)
         self._complete_events(execution_arn=execution_arn)
 
     def fail_execution(self, execution_arn: str, error: ErrorObject) -> None:
-        """Fail execution with error."""
+        """Fail execution with error (FAIL_WORKFLOW_EXECUTION decision)."""
         logger.exception("[%s] Completing execution with error.", execution_arn)
         execution: Execution = self._store.load(execution_arn=execution_arn)
-        execution.complete_fail(error=error)
+        execution.complete_fail(error=error)  # Sets CloseStatus.FAILED
         self._store.update(execution)
-        # set by complete_fail
         if execution.result is None:
             msg: str = "Execution result is required"
-
             raise IllegalStateException(msg)
         self._complete_events(execution_arn=execution_arn)
 
@@ -880,6 +826,19 @@ def on_failed(self, execution_arn: str, error: ErrorObject) -> None:
         """Fail execution. Observer method triggered by notifier."""
         self.fail_execution(execution_arn, error)
 
+    def on_timed_out(self, execution_arn: str, error: ErrorObject) -> None:
+        """Handle execution timeout (workflow timeout). Observer method triggered by notifier."""
+        logger.exception("[%s] Execution timed out.", execution_arn)
+        execution: Execution = self._store.load(execution_arn=execution_arn)
+        execution.complete_timeout(error=error)  # Sets CloseStatus.TIMED_OUT
+        self._store.update(execution)
+        self._complete_events(execution_arn=execution_arn)
+
+    def on_stopped(self, execution_arn: str, error: ErrorObject) -> None:
+        """Handle execution stop. Observer method triggered by notifier."""
+        # This should not be called directly - stop_execution handles termination
+        self.fail_execution(execution_arn, error)
+
     def on_wait_timer_scheduled(
         self, execution_arn: str, operation_id: str, delay: float
     ) -> None: