Add deterministic Python workflow UUIDv7 helper

Author: durable-workflow-ops

README.md

@@ -158,6 +158,7 @@ restart the worker process with a new id before serving changed workflow code.
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
 - **Payload-size warnings**: Structured warnings before oversized workflow, activity, schedule, signal, update, query, or search-attribute payloads reach the server
 - **Workflow definition guard**: Worker registration refuses same-id hot reloads when a workflow class definition changed
+- **Deterministic workflow helpers**: `ctx.now()`, `ctx.random()`, `ctx.uuid4()`, and `ctx.uuid7()` replay from workflow state
 - **Worker interceptors**: Typed hooks around workflow tasks, activity calls, and query tasks for tracing, logging, and custom metrics
 - **Metrics hooks**: Pluggable counters and histograms, with an optional Prometheus adapter
 

src/durable_workflow/workflow.py

@@ -9,10 +9,11 @@
 
 Determinism-sensitive helpers live on the :class:`WorkflowContext` passed to
 ``run``: :meth:`WorkflowContext.random`, :meth:`WorkflowContext.uuid4`,
-:meth:`WorkflowContext.now`, :meth:`WorkflowContext.patched`,
-:meth:`WorkflowContext.deprecate_patch`, and :meth:`WorkflowContext.side_effect` all
-produce values that are recorded on first execution and replayed verbatim
-on every subsequent replay of the same history.
+:meth:`WorkflowContext.uuid7`, :meth:`WorkflowContext.now`,
+:meth:`WorkflowContext.patched`, :meth:`WorkflowContext.deprecate_patch`, and
+:meth:`WorkflowContext.side_effect` all produce values that are recorded on
+first execution and replayed verbatim on every subsequent replay of the same
+history.
 """
 
 from __future__ import annotations
@@ -724,6 +725,7 @@ def __init__(self, *, run_id: str = "", current_time: datetime | None = None) ->
         self._current_time = current_time or datetime.now(timezone.utc)
         seed = int(hashlib.sha256(run_id.encode()).hexdigest()[:16], 16)
         self._rng = random.Random(seed)
+        self._uuid7_counter = 0
         self.logger = _ReplayLogger(_REPLAY_LOGGER)
 
     def schedule_activity(
@@ -871,6 +873,24 @@ def uuid4(self) -> uuid.UUID:
         rand_bytes = self._rng.getrandbits(128).to_bytes(16, "big")
         return uuid.UUID(bytes=rand_bytes, version=4)
 
+    def uuid7(self) -> uuid.UUID:
+        timestamp_ms = int(self._current_time.timestamp() * 1000)
+        if timestamp_ms < 0 or timestamp_ms >= (1 << 48):
+            raise ValueError("ctx.uuid7() requires ctx.now() within the UUIDv7 timestamp range")
+
+        rand_a = self._uuid7_counter & 0xFFF
+        self._uuid7_counter += 1
+        rand_b = self._rng.getrandbits(62)
+
+        value = (
+            (timestamp_ms << 80)
+            | (0x7 << 76)
+            | (rand_a << 64)
+            | (0x2 << 62)
+            | rand_b
+        )
+        return uuid.UUID(int=value)
+
 
 # ── Replay ───────────────────────────────────────────────────────────
 @dataclass

tests/test_replay.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import datetime, timezone
 
 from durable_workflow import serializer, workflow
 from durable_workflow.errors import ChildWorkflowFailed
@@ -376,7 +377,6 @@ def test_server_command_uses_payload_codec(self) -> None:
 
 class TestWorkflowContext:
     def test_now_returns_deterministic_time(self) -> None:
-        from datetime import datetime, timezone
         t = datetime(2026, 1, 1, tzinfo=timezone.utc)
         ctx = WorkflowContext(run_id="r1", current_time=t)
         assert ctx.now() == t
@@ -401,6 +401,34 @@ def test_uuid4_is_version_4(self) -> None:
         u = ctx.uuid4()
         assert u.version == 4
 
+    def test_uuid7_deterministic(self) -> None:
+        t = datetime(2026, 1, 1, 12, 30, 15, 123000, tzinfo=timezone.utc)
+        ctx1 = WorkflowContext(run_id="run-v7", current_time=t)
+        ctx2 = WorkflowContext(run_id="run-v7", current_time=t)
+
+        assert ctx1.uuid7() == ctx2.uuid7()
+        assert ctx1.uuid7() == ctx2.uuid7()
+
+    def test_uuid7_is_version_7_and_uses_context_time(self) -> None:
+        t = datetime(2026, 1, 1, 12, 30, 15, 123000, tzinfo=timezone.utc)
+        ctx = WorkflowContext(run_id="run-v7", current_time=t)
+
+        u = ctx.uuid7()
+
+        assert u.version == 7
+        assert (u.int >> 80) == int(t.timestamp() * 1000)
+
+    def test_uuid7_orders_same_tick_calls_by_sequence(self) -> None:
+        t = datetime(2026, 1, 1, 12, 30, 15, 123000, tzinfo=timezone.utc)
+        ctx = WorkflowContext(run_id="run-v7", current_time=t)
+
+        first = ctx.uuid7()
+        second = ctx.uuid7()
+
+        assert first < second
+        assert ((first.int >> 64) & 0xFFF) == 0
+        assert ((second.int >> 64) & 0xFFF) == 1
+
     def test_logger_silent_during_replay(self) -> None:
         import logging
         import logging.handlers