Skip to content

Expose a replay-safe new_guid on WorkflowContext #1202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
magic-peach wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Expose a replay-safe new_guid on WorkflowContext #1202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
91d5012
Expose a replay-safe new_guid on WorkflowContext
magic-peach Sep 9, 2026
0dfda55
Declare new_guid on the vendored OrchestrationContext interface
magic-peach Sep 11, 2026
67a2e95
Update dapr/ext/workflow/_durabletask/worker.py
magic-peach Sep 18, 2026
e75a3c0
Merge branch 'main' into feat/workflow-context-new-guid
sicoyle Sep 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions dapr/ext/workflow/_durabletask/task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
4 changes: 4 additions & 0 deletions dapr/ext/workflow/_durabletask/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1609,6 +1610,9 @@ def current_utc_datetime(self, value: datetime):
def is_replaying(self) -> bool:
return self._is_replaying

def new_guid(self) -> uuid.UUID:
return self.new_guid()
Comment on lines +1613 to +1614

@sicoyle sicoyle Sep 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this will recurse forever and break call_activity as is I believe. Can you delegate to the mixin's deterministic generator pls

@JeffreyJPZ JeffreyJPZ Sep 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, this one's my bad - should be super().new_guid()


def set_custom_status(self, custom_status: str) -> None:
if custom_status is not None and not isinstance(custom_status, str):
warnings.warn(
Expand Down
4 changes: 4 additions & 0 deletions dapr/ext/workflow/dapr_workflow_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -57,6 +58,9 @@ 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:
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)
Expand Down
17 changes: 17 additions & 0 deletions dapr/ext/workflow/workflow_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
7 changes: 7 additions & 0 deletions tests/ext/workflow/test_dapr_workflow_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import unittest
from datetime import datetime
from unittest import mock
from uuid import UUID

import dapr.ext.workflow._durabletask.internal.protos as pb
from dapr.ext.workflow import PropagatedHistory
Expand Down Expand Up @@ -54,6 +55,9 @@ def set_custom_status(self, custom_status):
def get_propagated_history(self):
return self._propagated_history

def new_guid(self):

@sicoyle sicoyle Sep 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this returns a hardcoded uuid, so nothing exercises the real implementation. Pls add something like:

def test_new_guid_is_deterministic_and_unique(self):
    ctx1 = worker._RuntimeOrchestrationContext('abc')
    ctx2 = worker._RuntimeOrchestrationContext('abc')
    g1 = [ctx1.new_guid() for _ in range(3)]
    g2 = [ctx2.new_guid() for _ in range(3)]
    assert all(isinstance(g, uuid.UUID) for g in g1)
    assert len(set(g1)) == 3   # unique within an execution
    assert g1 == g2            # stable across replays

return UUID('12345678-1234-5678-1234-567812345678')


class DaprWorkflowContextTest(unittest.TestCase):
def mock_client_activity(ctx: WorkflowActivityContext, input):
Expand Down Expand Up @@ -83,6 +87,9 @@ def test_workflow_context_functions(self):
dapr_wf_ctx.set_custom_status(mock_custom_status)
assert fakeContext.custom_status == mock_custom_status

new_guid_result = dapr_wf_ctx.new_guid()
assert new_guid_result == UUID('12345678-1234-5678-1234-567812345678')

def test_get_propagated_history_proxies_inner_context(self):
with mock.patch(
'dapr.ext.workflow._durabletask.worker._RuntimeOrchestrationContext',
Expand Down
Loading