feat(testing): add skip-time virtual clock

packages/aws-durable-execution-sdk-python-examples/test/conftest.py

@@ -230,8 +230,11 @@ def test_hello_world(durable_runner):
     else:
         if not handler:
             pytest.fail("handler is required for local mode tests")
+        skip_time = marker.kwargs.get("skip_time", True)
         # Create local runner (needs cleanup via context manager)
-        runner = DurableFunctionTestRunner(handler=handler, execution_timeout=60)
+        runner = DurableFunctionTestRunner(
+            handler=handler, execution_timeout=60, skip_time=skip_time
+        )
 
     # Wrap in adapter and use context manager for proper cleanup
     with TestRunnerAdapter(runner, runner_mode) as adapter:

packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py

@@ -34,10 +34,8 @@ def test_plugin(durable_runner):
     lambda_function_name="Plugin Wait",
 )
 def test_plugin_on_operation_end_called_for_wait_completed_during_suspend(
-    durable_runner, monkeypatch
+    durable_runner,
 ):
-    monkeypatch.setenv("DURABLE_EXECUTION_TIME_SCALE", "0.01")
-
     with durable_runner:
         result = durable_runner.run(input=None, timeout=30)
 

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py

@@ -20,6 +20,8 @@
 from aws_durable_execution_sdk_python_testing.token import CheckpointToken
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from aws_durable_execution_sdk_python.lambda_service import (
         Operation,
         OperationUpdate,
@@ -79,6 +81,7 @@ def apply(
         updates: list[OperationUpdate],
         client_token: str | None,
         dispatcher: CheckpointRequestDispatcher,
+        now: datetime,
     ) -> CheckpointResult:
         """Apply ``updates`` to ``execution`` and compute the response delta.
 
@@ -88,6 +91,10 @@ def apply(
         idempotency entry for a byte-identical replay of a retried call.
         The caller is responsible for the invocation gate, locking,
         persistence, and applying the returned effects.
+
+        ``now`` is the checkpoint's single source of "now", resolved from
+        the execution's clock, so all timestamps stamped by this apply
+        advance with modeled time under a skip clock.
         """
         effects: list[CheckpointEffect] = []
         if updates:
@@ -97,6 +104,7 @@ def apply(
                 updates=updates,
                 client_token=client_token,
                 touch=execution.touch_operation,
+                now=now,
             )
 
         new_token_sequence: int = execution.advance_token_sequence()

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py

@@ -20,6 +20,7 @@
 from aws_durable_execution_sdk_python_testing.checkpoint.transformer import (
     CheckpointRequestDispatcher,
 )
+from aws_durable_execution_sdk_python_testing.clock import RealClock
 from aws_durable_execution_sdk_python_testing.exceptions import (
     InvalidParameterValueException,
 )
@@ -28,8 +29,11 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_durable_execution_sdk_python.lambda_service import OperationUpdate
 
+    from aws_durable_execution_sdk_python_testing.clock import Clock
     from aws_durable_execution_sdk_python_testing.execution import Execution
     from aws_durable_execution_sdk_python_testing.observer import ExecutionObserver
     from aws_durable_execution_sdk_python_testing.scheduler import Scheduler
@@ -39,6 +43,11 @@
 # Canonical invocation-page byte budget, shared with the Executor.
 DEFAULT_MAX_INVOCATION_PAGE_BYTES = 5 * 1024 * 1024
 
+# Fallback clock used when no per-execution resolver is wired in. A real
+# clock keeps standalone construction (e.g. direct unit tests) on
+# wall-clock time.
+_REAL_CLOCK = RealClock()
+
 
 class CheckpointProcessor:
     """In-process checkpoint flow used by ``InMemoryServiceClient``.
@@ -52,10 +61,17 @@ def __init__(
         self,
         store: ExecutionStore,
         scheduler: Scheduler,  # noqa: ARG002 — kept for backward-compatible signature
+        clock_for: Callable[[str], Clock] | None = None,
     ):
         self._store = store
         self._observers: list[ExecutionObserver] = []
         self._dispatcher = CheckpointRequestDispatcher()
+        # Resolve the clock for an execution ARN so stamping shares the
+        # same "now" as the scheduler's readiness checks. Defaults to a
+        # real clock when no resolver is wired in.
+        self._clock_for: Callable[[str], Clock] = clock_for or (
+            lambda _arn: _REAL_CLOCK
+        )
 
     def add_execution_observer(self, observer: ExecutionObserver) -> None:
         """Add observer for execution events."""
@@ -91,12 +107,14 @@ def process_checkpoint(
             msg = "Invalid checkpoint token"
             raise InvalidParameterValueException(msg)
 
+        now = self._clock_for(token.execution_arn).now()
         result = CheckpointCore.apply(
             execution,
             checkpoint_token,
             updates,
             client_token,
             self._dispatcher,
+            now,
         )
 
         self._store.update(execution)

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py

@@ -34,22 +34,29 @@ def process(
         current_op: Operation | None,
         notifier: ExecutionNotifier,
         execution_arn: str,
+        now: datetime.datetime,
     ) -> Operation | None:
-        """Process an operation update and return the transformed operation."""
+        """Process an operation update and return the transformed operation.
+
+        ``now`` is the checkpoint's single source of "now", resolved from
+        the execution's clock so timestamps advance with modeled time
+        under a skip clock and match wall-clock under a real clock.
+        """
         raise NotImplementedError
 
     def _get_start_time(
-        self, current_operation: Operation | None
+        self, current_operation: Operation | None, now: datetime.datetime
     ) -> datetime.datetime | None:
         start_time: datetime.datetime | None = (
-            current_operation.start_timestamp
-            if current_operation
-            else datetime.datetime.now(tz=datetime.UTC)
+            current_operation.start_timestamp if current_operation else now
         )
         return start_time
 
     def _get_end_time(
-        self, current_operation: Operation | None, status: OperationStatus
+        self,
+        current_operation: Operation | None,
+        status: OperationStatus,
+        now: datetime.datetime,
     ) -> datetime.datetime | None:
         """Get end timestamp for operation based on current state and status."""
         if current_operation and current_operation.end_timestamp:
@@ -61,7 +68,7 @@ def _get_end_time(
             OperationStatus.TIMED_OUT,
             OperationStatus.STOPPED,
         }:
-            return datetime.datetime.now(tz=datetime.UTC)
+            return now
         return None
 
     def _create_execution_details(
@@ -151,19 +158,22 @@ def _translate_update_to_operation(
         update: OperationUpdate,
         current_operation: Operation | None,
         status: OperationStatus,
+        now: datetime.datetime,
     ) -> Operation:
         """Transform OperationUpdate to Operation, always creating new Operation."""
-        start_time: datetime.datetime | None = self._get_start_time(current_operation)
+        start_time: datetime.datetime | None = self._get_start_time(
+            current_operation, now
+        )
         end_time: datetime.datetime | None = self._get_end_time(
-            current_operation, status
+            current_operation, status, now
         )
 
         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)
         invoke_details = self._create_invoke_details(update)
-        wait_details = self._create_wait_details(update, current_operation)
+        wait_details = self._create_wait_details(update, current_operation, now)
 
         return Operation(
             operation_id=update.operation_id,
@@ -183,7 +193,10 @@ def _translate_update_to_operation(
         )
 
     def _create_wait_details(
-        self, update: OperationUpdate, current_operation: Operation | None
+        self,
+        update: OperationUpdate,
+        current_operation: Operation | None,
+        now: datetime.datetime,
     ) -> WaitDetails | None:
         """Create WaitDetails from OperationUpdate."""
         if update.operation_type == OperationType.WAIT and update.wait_options:
@@ -192,8 +205,8 @@ def _create_wait_details(
                     current_operation.wait_details.scheduled_end_timestamp
                 )
             else:
-                scheduled_end_timestamp = datetime.datetime.now(
-                    tz=datetime.UTC
-                ) + timedelta(seconds=update.wait_options.wait_seconds)
+                scheduled_end_timestamp = now + timedelta(
+                    seconds=update.wait_options.wait_seconds
+                )
             return WaitDetails(scheduled_end_timestamp=scheduled_end_timestamp)
         return None

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py

@@ -35,6 +35,7 @@ def process(
         current_op: Operation | None,
         notifier: ExecutionNotifier,  # noqa: ARG002
         execution_arn: str,  # noqa: ARG002
+        now: datetime.datetime,
     ) -> Operation:
         """Process CALLBACK operation update with scheduler integration for activities."""
         match update.action:
@@ -58,10 +59,12 @@ def process(
 
                 status: OperationStatus = OperationStatus.STARTED
 
-                start_time: datetime.datetime | None = self._get_start_time(current_op)
+                start_time: datetime.datetime | None = self._get_start_time(
+                    current_op, now
+                )
 
                 end_time: datetime.datetime | None = self._get_end_time(
-                    current_op, status
+                    current_op, status, now
                 )
 
                 operation: Operation = Operation(

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py

@@ -20,6 +20,8 @@
 
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier
 
 
@@ -32,6 +34,7 @@ def process(
         current_op: Operation | None,
         notifier: ExecutionNotifier,  # noqa: ARG002
         execution_arn: str,  # noqa: ARG002
+        now: datetime,
     ) -> Operation:
         """Process CONTEXT operation update for context state transitions."""
         match update.action:
@@ -41,18 +44,21 @@ def process(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.STARTED,
+                    now=now,
                 )
             case OperationAction.SUCCEED:
                 return self._translate_update_to_operation(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.SUCCEEDED,
+                    now=now,
                 )
             case OperationAction.FAIL:
                 return self._translate_update_to_operation(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.FAILED,
+                    now=now,
                 )
             case _:
                 msg: str = "Invalid action for CONTEXT operation."

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py

@@ -17,6 +17,8 @@
 
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier
 
 
@@ -29,6 +31,7 @@ def process(
         current_op: Operation | None,  # noqa: ARG002
         notifier: ExecutionNotifier,
         execution_arn: str,
+        now: datetime,  # noqa: ARG002
     ) -> Operation | None:
         """Process EXECUTION operation update for workflow completion/failure."""
         match update.action:

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta
 from typing import TYPE_CHECKING
 
 from aws_durable_execution_sdk_python.lambda_service import (
@@ -34,6 +34,7 @@ def process(
         current_op: Operation | None,
         notifier: ExecutionNotifier,
         execution_arn: str,
+        now: datetime,
     ) -> Operation:
         """Process STEP operation update with scheduler integration for retries."""
         match update.action:
@@ -42,6 +43,7 @@ def process(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.STARTED,
+                    now=now,
                 )
             case OperationAction.RETRY:
                 # set Status=PENDING, next attempt time, attempt count + 1
@@ -50,7 +52,7 @@ def process(
                     if update.step_options
                     else 0
                 )
-                next_attempt_time = datetime.now(UTC) + timedelta(seconds=delay)
+                next_attempt_time = now + timedelta(seconds=delay)
 
                 # Build new step_details with incremented attempt
                 current_attempt = (
@@ -72,9 +74,7 @@ def process(
                     status=OperationStatus.PENDING,
                     parent_id=update.parent_id,
                     name=update.name,
-                    start_timestamp=(
-                        current_op.start_timestamp if current_op else datetime.now(UTC)
-                    ),
+                    start_timestamp=(current_op.start_timestamp if current_op else now),
                     end_timestamp=None,
                     sub_type=update.sub_type,
                     execution_details=current_op.execution_details
@@ -103,12 +103,14 @@ def process(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.SUCCEEDED,
+                    now=now,
                 )
             case OperationAction.FAIL:
                 return self._translate_update_to_operation(
                     update=update,
                     current_operation=current_op,
                     status=OperationStatus.FAILED,
+                    now=now,
                 )
             case _:
                 msg: str = "Invalid action for STEP operation."