feat(testing): add skip-time virtual clock

- Add Clock protocol with RealClock and SkipClock implementations
- Give each ExecutionWorker its own clock via the registry factory
- Drive wait and step-retry timers off the clock so modeled delays
  complete instantly under skip while history keeps real durations
- Default DurableFunctionTestRunner to skip_time=True; pass
  skip_time=False for real wall-clock timing
- Remove the DURABLE_EXECUTION_TIME_SCALE env-var scaling

Author: yaythomas

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/processors/wait.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-import logging
-import os
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING
 
@@ -43,12 +41,9 @@ def process(
                 wait_seconds = (
                     update.wait_options.wait_seconds if update.wait_options else 0
                 )
-                time_scale = float(os.getenv("DURABLE_EXECUTION_TIME_SCALE", "1.0"))
-                logging.info("Using DURABLE_EXECUTION_TIME_SCALE: %f", time_scale)
-                scaled_wait_seconds = wait_seconds * time_scale
 
                 scheduled_end_timestamp = datetime.now(UTC) + timedelta(
-                    seconds=scaled_wait_seconds
+                    seconds=wait_seconds
                 )
 
                 # Create WaitDetails with scheduled timestamp

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

@@ -0,0 +1,63 @@
+"""Clock sources for scheduler-driven durable timers.
+
+A clock decides how long the local runner actually waits before a
+scheduled wake fires. History timestamps are written from the real
+modeled durations regardless of the clock, so a :class:`SkipClock` run
+records the same event history as a :class:`RealClock` run and only the
+wall-clock spent waiting differs.
+"""
+
+from __future__ import annotations
+
+import datetime
+import threading
+from typing import Protocol
+
+
+class Clock(Protocol):
+    """Time source used to arm and evaluate durable timers."""
+
+    def now(self) -> datetime.datetime:
+        """Return the current time on this clock."""
+        ...
+
+    def arm(self, wake: datetime.datetime) -> float:
+        """Return the real seconds to wait before firing at ``wake``."""
+        ...
+
+
+class RealClock:
+    """Wall-clock time source. Timers wait the real modeled duration."""
+
+    def now(self) -> datetime.datetime:
+        return datetime.datetime.now(datetime.UTC)
+
+    def arm(self, wake: datetime.datetime) -> float:
+        return max((wake - self.now()).total_seconds(), 0.0)
+
+
+class SkipClock:
+    """Virtual clock that jumps to each armed timer horizon.
+
+    Mutable owner of one execution's virtual now. Arming a wake advances
+    virtual now to that horizon and returns a zero delay, so the wake
+    fires on the next event-loop turn rather than after the modeled
+    duration. Guards its state with a lock because ``arm`` runs on the
+    scheduling thread while ``now`` runs on the fire worker thread.
+    """
+
+    def __init__(self) -> None:
+        self._now: datetime.datetime | None = None
+        self._lock: threading.Lock = threading.Lock()
+
+    def now(self) -> datetime.datetime:
+        with self._lock:
+            if self._now is None:
+                self._now = datetime.datetime.now(datetime.UTC)
+            return self._now
+
+    def arm(self, wake: datetime.datetime) -> float:
+        with self._lock:
+            if self._now is None or wake > self._now:
+                self._now = wake
+        return 0.0

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

@@ -1033,8 +1033,7 @@ def _schedule_earliest_pending(self, execution_arn: str) -> None:
                 existing.cancel()
             return
 
-        now = datetime.now(UTC)
-        delay = max((earliest - now).total_seconds(), 0.0)
+        delay = self._registry.get_or_create(execution_arn).clock.arm(earliest)
 
         completion_event = self._completion_events.get(execution_arn)
         # Cancel-then-arm atomically under the supervisor lock so
@@ -1064,7 +1063,7 @@ def _fire_due_operations(self, execution_arn: str) -> bool:
         if execution.is_complete:
             return False
 
-        now = datetime.now(UTC)
+        now = self._registry.get_or_create(execution_arn).clock.now()
         completed_any = False
         for op in list(execution.operations):
             if (

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

@@ -38,6 +38,7 @@
     DEFAULT_MAX_INVOCATION_PAGE_BYTES,
 )
 from aws_durable_execution_sdk_python_testing.client import InMemoryServiceClient
+from aws_durable_execution_sdk_python_testing.clock import RealClock, SkipClock
 from aws_durable_execution_sdk_python_testing.worker.registry import ExecutionRegistry
 from aws_durable_execution_sdk_python_testing.exceptions import (
     DurableFunctionsLocalRunnerError,
@@ -605,6 +606,7 @@ def __init__(
         execution_timeout: int = 300,
         invocation_timeout: int = 900,
         store: ExecutionStore | None = None,
+        skip_time: bool = True,  # noqa: FBT001, FBT002
     ):
         self._execution_timeout = execution_timeout
         self._invocation_timeout = invocation_timeout
@@ -625,7 +627,11 @@ def __init__(
             store=self._store,
             scheduler=self._scheduler,
         )
-        self._registry = ExecutionRegistry(self._store, self._scheduler)
+        self._registry = ExecutionRegistry(
+            self._store,
+            self._scheduler,
+            clock_factory=SkipClock if skip_time else RealClock,
+        )
         self._service_client = InMemoryServiceClient(
             self._checkpoint_processor, self._registry
         )

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/execution_worker.py

@@ -10,6 +10,7 @@
 if TYPE_CHECKING:
     from concurrent.futures import Future
 
+    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.scheduler import Scheduler
     from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore
@@ -44,12 +45,14 @@ def __init__(
         scheduler: Scheduler,
         registry: ExecutionRegistry,
         lane: SerialTaskLane,
+        clock: Clock,
     ) -> None:
         self._arn: str = execution_arn
         self._store: ExecutionStore = store
         self._scheduler: Scheduler = scheduler
         self._registry: ExecutionRegistry = registry
         self._lane: SerialTaskLane = lane
+        self._clock: Clock = clock
         self._status: InvocationState = InvocationState.PRE_INVOKE
 
     @classmethod
@@ -59,12 +62,18 @@ def create(
         store: ExecutionStore,
         scheduler: Scheduler,
         registry: ExecutionRegistry,
+        clock: Clock,
     ) -> ExecutionWorker:
         """Build a worker and start its lane."""
         lane: SerialTaskLane = SerialTaskLane.create(
             name=f"execution-worker-{execution_arn}"
         )
-        return cls(execution_arn, store, scheduler, registry, lane)
+        return cls(execution_arn, store, scheduler, registry, lane, clock)
+
+    @property
+    def clock(self) -> Clock:
+        """The clock arming and evaluating this execution's timers."""
+        return self._clock
 
     @property
     def status(self) -> InvocationState:

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/registry.py

@@ -5,11 +5,14 @@
 import threading
 from typing import TYPE_CHECKING
 
+from aws_durable_execution_sdk_python_testing.clock import Clock, RealClock
 from aws_durable_execution_sdk_python_testing.worker.execution_worker import (
     ExecutionWorker,
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_durable_execution_sdk_python_testing.scheduler import Scheduler
     from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore
 
@@ -24,9 +27,15 @@ class ExecutionRegistry:
     a given ARN.
     """
 
-    def __init__(self, store: ExecutionStore, scheduler: Scheduler) -> None:
+    def __init__(
+        self,
+        store: ExecutionStore,
+        scheduler: Scheduler,
+        clock_factory: Callable[[], Clock] = RealClock,
+    ) -> None:
         self._store = store
         self._scheduler = scheduler
+        self._clock_factory = clock_factory
         self._workers: dict[str, ExecutionWorker] = {}
         self._lock = threading.Lock()
 
@@ -36,7 +45,11 @@ def get_or_create(self, execution_arn: str) -> ExecutionWorker:
             worker: ExecutionWorker | None = self._workers.get(execution_arn)
             if worker is None:
                 worker = ExecutionWorker.create(
-                    execution_arn, self._store, self._scheduler, self
+                    execution_arn,
+                    self._store,
+                    self._scheduler,
+                    self,
+                    self._clock_factory(),
                 )
                 self._workers[execution_arn] = worker
             return worker

packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py

@@ -0,0 +1,75 @@
+"""Tests for the durable timer clocks."""
+
+from __future__ import annotations
+
+import datetime
+from unittest.mock import Mock
+
+from aws_durable_execution_sdk_python_testing.clock import (
+    RealClock,
+    SkipClock,
+)
+from aws_durable_execution_sdk_python_testing.worker.registry import (
+    ExecutionRegistry,
+)
+
+
+def test_real_clock_arm_returns_remaining_wall_seconds() -> None:
+    clock = RealClock()
+    wake = clock.now() + datetime.timedelta(seconds=30)
+
+    delay: float = clock.arm(wake)
+
+    # The real clock waits the modeled duration, minus the tiny elapsed
+    # time between now() calls.
+    assert 29.0 <= delay <= 30.0
+
+
+def test_real_clock_arm_never_negative_for_past_wake() -> None:
+    clock = RealClock()
+    past = clock.now() - datetime.timedelta(seconds=30)
+
+    assert clock.arm(past) == 0.0
+
+
+def test_skip_clock_arm_returns_zero_delay() -> None:
+    clock = SkipClock()
+    wake = clock.now() + datetime.timedelta(seconds=300)
+
+    assert clock.arm(wake) == 0.0
+
+
+def test_skip_clock_now_jumps_to_armed_horizon() -> None:
+    clock = SkipClock()
+    wake = clock.now() + datetime.timedelta(seconds=300)
+
+    clock.arm(wake)
+
+    assert clock.now() == wake
+
+
+def test_skip_clock_is_monotonic() -> None:
+    clock = SkipClock()
+    start = clock.now()
+    far = start + datetime.timedelta(seconds=300)
+    near = start + datetime.timedelta(seconds=10)
+
+    clock.arm(far)
+    clock.arm(near)
+
+    # A later arm for an earlier horizon must not move virtual time back.
+    assert clock.now() == far
+
+
+def test_registry_uses_clock_factory() -> None:
+    skip_registry = ExecutionRegistry(Mock(), Mock(), clock_factory=SkipClock)
+    real_registry = ExecutionRegistry(Mock(), Mock(), clock_factory=RealClock)
+
+    assert isinstance(skip_registry.get_or_create("arn").clock, SkipClock)
+    assert isinstance(real_registry.get_or_create("arn").clock, RealClock)
+
+
+def test_registry_defaults_to_real_clock() -> None:
+    registry = ExecutionRegistry(Mock(), Mock())
+
+    assert isinstance(registry.get_or_create("arn").clock, RealClock)

packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py

@@ -0,0 +1,50 @@
+"""End-to-end tests for skip-time behaviour on the local runner."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from aws_durable_execution_sdk_python.config import Duration
+from aws_durable_execution_sdk_python.context import DurableContext
+from aws_durable_execution_sdk_python.execution import (
+    InvocationStatus,
+    durable_execution,
+)
+
+from aws_durable_execution_sdk_python_testing.runner import (
+    DurableFunctionTestRunner,
+)
+
+
+_MODELED_WAIT_SECONDS = 300
+
+
+@durable_execution
+def _waits_then_returns(event: Any, context: DurableContext) -> str:  # noqa: ARG001
+    context.wait(Duration.from_seconds(_MODELED_WAIT_SECONDS))
+    return "done"
+
+
+def test_skip_time_completes_long_wait_and_keeps_faithful_history() -> None:
+    # skip_time defaults to True. A modeled 300s wait must complete well
+    # within the 30s execution timeout, which is only possible if the
+    # runner does not spend the modeled duration in wall-clock time.
+    with DurableFunctionTestRunner(handler=_waits_then_returns) as runner:
+        result = runner.run(input="x", execution_timeout=30)
+
+    assert result.status is InvocationStatus.SUCCEEDED
+
+    wait_ops = [op for op in result.operations if op.operation_type.value == "WAIT"]
+    assert len(wait_ops) == 1
+    wait_op = wait_ops[0]
+
+    assert wait_op.status.value == "SUCCEEDED"
+    assert wait_op.start_timestamp is not None
+    assert wait_op.scheduled_end_timestamp is not None
+
+    # History records the real modeled duration, not the skipped
+    # wall-clock time.
+    modeled = (
+        wait_op.scheduled_end_timestamp - wait_op.start_timestamp
+    ).total_seconds()
+    assert abs(modeled - _MODELED_WAIT_SECONDS) < 1.0