feat(workflow): add new_guid to DaprWorkflowContext

dapr/ext/workflow/AGENTS.md

@@ -168,6 +168,7 @@ Passed to workflow functions as the first argument:
 - `create_timer(fire_at)` → `Task` (accepts `datetime` or `timedelta`)
 - `wait_for_external_event(name)` → `Task`
 - `set_custom_status(status)` / `continue_as_new(new_input, *, save_events)`
+- `new_guid()` → `uuid.UUID` — deterministic across workflow replays
 
 Module-level functions:
 - `when_all(tasks)` → `WhenAllTask` — wait for all tasks to complete

dapr/ext/workflow/_durabletask/task.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import math
+import uuid
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import Any, Callable, Generator, Generic, Optional, TypeVar, Union
@@ -89,6 +90,17 @@ def set_custom_status(self, custom_status: str) -> None:
         """
         pass
 
+    @abstractmethod
+    def new_guid(self) -> uuid.UUID:
+        """Create a new GUID that is safe for replay within an orchestration.
+
+        Returns
+        -------
+        uuid.UUID
+            A new, deterministically generated GUID.
+        """
+        pass
+
     @abstractmethod
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task:
         """Create a Timer Task to fire after at the specified deadline.

dapr/ext/workflow/_durabletask/worker.py

@@ -17,6 +17,7 @@
 import random
 import threading
 import time
+import uuid
 import warnings
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta, timezone
@@ -1729,6 +1730,12 @@ def current_utc_datetime(self, value: datetime):
     def is_replaying(self) -> bool:
         return self._is_replaying
 
+    def new_guid(self) -> uuid.UUID:
+        # Call the mixin implementation directly because ``super().new_guid()``
+        # first resolves to the abstract method ``OrchestrationContext.new_guid()``,
+        # which returns ``None``.
+        return deterministic.DeterministicContextMixin.new_guid(self)
+
     def set_custom_status(self, custom_status: str) -> None:
         if custom_status is not None and not isinstance(custom_status, str):
             warnings.warn(

dapr/ext/workflow/dapr_workflow_context.py

@@ -15,6 +15,7 @@
 
 from datetime import datetime, timedelta
 from typing import Any, Callable, List, Optional, TypeVar, Union
+from uuid import UUID
 
 from dapr.ext.workflow._durabletask import task
 from dapr.ext.workflow.logger import Logger, LoggerOptions
@@ -57,6 +58,10 @@ def set_custom_status(self, custom_status: str) -> None:
         self._logger.debug(f'{self.instance_id}: Setting custom status to {custom_status}')
         self.__obj.set_custom_status(custom_status)
 
+    def new_guid(self) -> UUID:
+        self._logger.debug(f'{self.instance_id}: Generating new GUID')
+        return self.__obj.new_guid()
+
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         self._logger.debug(f'{self.instance_id}: Creating timer to fire at {fire_at} time')
         return self.__obj.create_timer(fire_at)

dapr/ext/workflow/workflow_context.py

@@ -18,6 +18,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import Any, Callable, Generator, Optional, TypeVar, Union
+from uuid import UUID
 
 from dapr.ext.workflow._durabletask import task
 from dapr.ext.workflow.propagation import PropagatedHistory, PropagationScope
@@ -90,6 +91,22 @@ def set_custom_status(self, custom_status: str) -> None:
         """Set the custom status."""
         pass
 
+    @abstractmethod
+    def new_guid(self) -> UUID:
+        """Create a new GUID that is safe for replay within a workflow.
+
+        The GUID is deterministically derived from the workflow instance ID,
+        the current replay-safe time, and a counter that increments on each
+        call, so calling this repeatedly returns different values within a
+        single execution while remaining stable across replays.
+
+        Returns
+        -------
+        uuid.UUID
+            A replay-safe, deterministically generated GUID.
+        """
+        pass
+
     @abstractmethod
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         """Create a Timer Task to fire after at the specified deadline.

tests/ext/workflow/test_dapr_workflow_context.py

@@ -14,6 +14,7 @@
 """
 
 import unittest
+import uuid
 from datetime import datetime
 from unittest import mock
 
@@ -120,3 +121,15 @@ def test_get_propagated_history_returns_none_when_not_set(self):
             fake = worker._RuntimeOrchestrationContext(mock_instance_id)
             dapr_wf_ctx = DaprWorkflowContext(fake)
             assert dapr_wf_ctx.get_propagated_history() is None
+
+    def test_new_guid_is_unique_within_execution_and_deterministic_across_replays(self):
+        instance_id = 'abc'
+        ctx1 = DaprWorkflowContext(worker._RuntimeOrchestrationContext(instance_id))
+        ctx2 = DaprWorkflowContext(worker._RuntimeOrchestrationContext(instance_id))
+        guid_count = 3
+        g1 = [ctx1.new_guid() for _ in range(guid_count)]
+        g2 = [ctx2.new_guid() for _ in range(guid_count)]
+
+        assert all(isinstance(g, uuid.UUID) for g in g1)
+        assert len(set(g1)) == guid_count  # unique within an execution
+        assert g1 == g2  # deterministic across replays

tests/integration/test_workflow_replay_safety.py

@@ -0,0 +1,127 @@
+# Copyright 2026 The Dapr Authors
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""End-to-end verification of replay-safe workflow context behavior.
+
+These tests run workflows through a real Dapr sidecar. After an activity
+completes, the workflow is resumed by replaying its history, so the replay
+assertions exercise ``DaprWorkflowContext`` rather than only the underlying
+deterministic context.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from uuid import UUID
+
+import pytest
+
+HOST = '127.0.0.1'
+GRPC_PORT = '13501'
+COMPLETION_TIMEOUT_S = 60
+GUID_COUNT = 3
+
+
+@pytest.fixture(scope='module')
+def sidecar(dapr_env):
+    """Start the Dapr sidecar used by the workflow replay tests."""
+    return dapr_env.start_sidecar(app_id='test-workflow-replay-safety')
+
+
+def _run_workflow(workflow, *, activities: tuple[Any, ...] = ()) -> dict[str, Any] | list[str]:
+    """Run a workflow to completion through a real Dapr sidecar and decode its JSON output."""
+    from dapr.ext.workflow import DaprWorkflowClient, WorkflowRuntime
+
+    runtime = WorkflowRuntime(host=HOST, port=GRPC_PORT)
+    runtime.register_workflow(workflow)
+    for activity in activities:
+        runtime.register_activity(activity)
+
+    client = DaprWorkflowClient(host=HOST, port=GRPC_PORT)
+    runtime.start()
+    try:
+        instance_id = client.schedule_new_workflow(workflow)
+        state = client.wait_for_workflow_completion(
+            instance_id, timeout_in_seconds=COMPLETION_TIMEOUT_S
+        )
+    finally:
+        client.close()
+        runtime.shutdown()
+
+    assert state is not None, 'workflow did not reach a terminal state'
+    assert state.runtime_status.name == 'COMPLETED'
+    return json.loads(state.serialized_output)
+
+
+def test_new_guid_generates_unique_guids_within_execution(sidecar):
+    """Each call to ``new_guid()`` produces a unique GUID within one execution."""
+    from dapr.ext.workflow import DaprWorkflowContext, WorkflowActivityContext
+
+    def echo_guids(ctx: WorkflowActivityContext, values: list[str]) -> list[str]:
+        return values
+
+    def generate_guids(ctx: DaprWorkflowContext, _input: None):
+        generated_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)]
+        yield ctx.call_activity(echo_guids, input=generated_guids)
+        return generated_guids
+
+    generated_guids = _run_workflow(generate_guids, activities=(echo_guids,))
+
+    assert isinstance(generated_guids, list)
+
+    # Validate serialized GUIDs
+    for value in generated_guids:
+        UUID(value)
+
+    assert len(set(generated_guids)) == len(generated_guids)
+
+
+def test_new_guid_is_deterministic_across_replays(sidecar):
+    """GUID sequences remain unchanged when the workflow replays its history.
+
+    The echo activity returns the GUIDS from the initial execution.
+    During replay, that result comes from history instead of executing the
+    activity again, so comparing the generated and echoed values detects if replay
+    produced different GUIDs than the initial execution.
+    """
+    from dapr.ext.workflow import DaprWorkflowContext, WorkflowActivityContext
+
+    def echo_guids(ctx: WorkflowActivityContext, values: list[str]) -> list[str]:
+        return values
+
+    def generate_guids(ctx: DaprWorkflowContext, _input: None):
+        first_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)]
+        first_echo = yield ctx.call_activity(echo_guids, input=first_guids)
+
+        second_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)]
+        second_echo = yield ctx.call_activity(echo_guids, input=second_guids)
+
+        return {
+            'first_guids': first_guids,
+            'first_echo': first_echo,
+            'second_guids': second_guids,
+            'second_echo': second_echo,
+        }
+
+    result = _run_workflow(generate_guids, activities=(echo_guids,))
+
+    assert isinstance(result, dict)
+    assert result['first_guids'] == result['first_echo']
+    assert result['second_guids'] == result['second_echo']
+
+    # Validate serialized GUIDs
+    for values in (result['first_guids'], result['second_guids']):
+        for value in values:
+            UUID(value)
+
+    all_guids = result['first_guids'] + result['second_guids']
+    assert len(set(all_guids)) == len(all_guids)