Workflow versioning

Signed-off-by: Albert Callarisa <albert@diagrid.io>

Author: acroca

dev-requirements.txt

@@ -14,7 +14,7 @@ Flask>=1.1
 # needed for auto fix
 ruff===0.14.1
 # needed for dapr-ext-workflow
-durabletask-dapr >= 0.2.0a15
+durabletask-dapr >= 0.2.0a16
 # needed for .env file loading in examples
 python-dotenv>=1.0.0
 # needed for enhanced schema generation from function features

examples/workflow/README.md

@@ -461,3 +461,50 @@ app1 - received workflow error from app2
 ```
 among others. This shows that the workflow calls are failing as expected, and they are being handled as expected too.
 
+
+### Versioning
+
+This example demonstrates how to version a workflow.
+The test consists of two parts:
+1. Uses most of the common features of the workflow versioning. It also leaves some workflows stalled to demonstrate the stalled workflow feature.
+2. Fixes the stalled workflows to get them to completion.
+
+It had to be done in two parts because the runtime needs to be restarted in order to rerun stalled workflows.
+
+ The Dapr CLI can be started using the following command:
+
+<!--STEP
+name: Run the versioning example
+match_order: none
+expected_stdout_lines:
+  - "== APP == test1: triggering workflow"
+  - "== APP == test1: Received workflow call for version1"
+  - "== APP == test1: Finished workflow for version1"
+  - "== APP == test2: triggering workflow"
+  - "== APP == test2: Received workflow call for version1"
+  - "== APP == test2: Finished workflow for version1"
+  - "== APP == test3: triggering workflow"
+  - "== APP == test3: Received workflow call for version2"
+  - "== APP == test3: Finished workflow for version2"
+  - "== APP == test4: start"
+  - "== APP == test4: patch1 is patched"
+  - "== APP == test5: start"
+  - "== APP == test5: patch1 is not patched"
+  - "== APP == test5: patch2 is patched"
+  - "== APP == test6: start"
+  - "== APP == test6: patch1 is patched"
+  - "== APP == test6: patch2 is patched"
+  - "== APP == test7: Received workflow call for version1"
+  - "== APP == test7: Workflow is stalled"
+  - "== APP == test8: Workflow is stalled"
+  - "== APP == test100: part2"
+  - "== APP == test100: Finished stalled version1 workflow"
+  - "== APP == test100: Finished stalled patching workflow"
+timeout_seconds: 60
+-->
+
+```sh
+dapr run --app-id wf-versioning-example -- python3 versioning.py part1
+dapr run --app-id wf-versioning-example --log-level debug -- python3 versioning.py part2
+```
+<!--END_STEP-->

examples/workflow/versioning.py

@@ -0,0 +1,280 @@
+# -*- coding: utf-8 -*-
+# 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.
+
+import sys
+import time
+
+import dapr.ext.workflow as wf
+from durabletask.task import Activity
+
+
+current_test = 0
+
+def print_test(message):
+    print(f'test{current_test}: {message}', flush=True)
+
+print_activity = None
+
+wfr = None
+def new_wfr():
+    global wfr
+    global print_activity
+
+    if wfr is not None:
+        wfr.shutdown()
+    wfr = wf.WorkflowRuntime()
+    print_activity = lambda ctx, input: print_test(input)
+    wfr.register_activity(print_activity, name='print_activity')
+
+new_wfr()
+
+def test_full_versioning(client: wf.DaprWorkflowClient):
+    global current_test
+
+    # Start with only one version defined. Runnig the workflow should run this version as it normally would.
+    current_test = 1
+    @wfr.versioned_workflow(name="workflow", is_latest=True)
+    def version1_workflow(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='Received workflow call for version1')
+        yield ctx.wait_for_external_event(name='event')
+        yield ctx.call_activity(print_activity, input='Finished workflow for version1')
+        return 1
+
+    print_test('triggering workflow')
+    instance_id = client.schedule_new_workflow(workflow=version1_workflow)
+    client.raise_workflow_event(instance_id, event_name='event')
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+    # Now we start a workflow, but introduce a latest version half way. It should resume the execution in the old version.
+    current_test = 2
+    print_test('triggering workflow')
+    instance_id = client.schedule_new_workflow(workflow=version1_workflow)
+    time.sleep(2) # wait for the workflow to start and wait for the event
+
+    @wfr.versioned_workflow(name="workflow", is_latest=True)
+    def version2_workflow(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='Received workflow call for version2')
+        yield ctx.wait_for_external_event(name='event')
+        yield ctx.call_activity(print_activity, input='Finished workflow for version2')
+        return 1
+
+    client.raise_workflow_event(instance_id, event_name='event')
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+
+    # Now we have the two versions defined, running the workflow now should run v2 as it's the latest version.
+    current_test = 3
+    print_test('triggering workflow')
+    instance_id = client.schedule_new_workflow(workflow=version1_workflow)
+    client.raise_workflow_event(instance_id, event_name='event')
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+def test_patching(client: wf.DaprWorkflowClient):
+    global current_test
+
+    @wfr.workflow
+    def patching_workflow(ctx: wf.DaprWorkflowContext):
+        # This function will be changed throughout the test, to simulate different scenarios
+        return workflow_code(ctx)
+
+
+    # Runs the patched branch by default
+    current_test = 4
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='start')
+        if ctx.is_patched('patch1'):
+            yield ctx.call_activity(print_activity, input='patch1 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch1 is not patched')
+        return 1
+
+    instance_id = client.schedule_new_workflow(workflow=patching_workflow)
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+    # When the execution passed the place where a patch is introduced, it should be not patched.
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='start')
+        yield ctx.wait_for_external_event(name='event')
+        if ctx.is_patched('patch2'):
+            yield ctx.call_activity(print_activity, input='patch2 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch2 is not patched')
+        return 1
+
+    current_test = 5
+    instance_id = client.schedule_new_workflow(workflow=patching_workflow)
+    time.sleep(2)
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='start')
+        if ctx.is_patched('patch1'):
+            yield ctx.call_activity(print_activity, input='patch1 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch1 is not patched')
+        yield ctx.wait_for_external_event(name='event')
+        if ctx.is_patched('patch2'):
+            yield ctx.call_activity(print_activity, input='patch2 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch2 is not patched')
+        return 1
+
+    client.raise_workflow_event(instance_id, event_name='event')
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+    # It remembers previous patches.
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='start')
+        if ctx.is_patched('patch1'):
+            pass # keep it silenced for now, we'll add logs later and this ones would confuse the test
+        else:
+            pass
+        yield ctx.wait_for_external_event(name='event')
+        if ctx.is_patched('patch2'):
+            yield ctx.call_activity(print_activity, input='patch2 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch2 is not patched')
+        return 1
+
+    current_test = 6
+    instance_id = client.schedule_new_workflow(workflow=patching_workflow)
+    time.sleep(2)
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='start')
+        if ctx.is_patched('patch1'):
+            yield ctx.call_activity(print_activity, input='patch1 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch1 is not patched')
+        yield ctx.wait_for_external_event(name='event')
+        if ctx.is_patched('patch2'):
+            yield ctx.call_activity(print_activity, input='patch2 is patched')
+        else:
+            yield ctx.call_activity(print_activity, input='patch2 is not patched')
+        return 1
+
+    client.raise_workflow_event(instance_id, event_name='event')
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
+
+
+def test_full_versioning_stall(client: wf.DaprWorkflowClient):
+    global current_test
+
+    new_wfr()
+    @wfr.versioned_workflow(name="stall_workflow", is_latest=True)
+    def version1_workflow(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='Received workflow call for version1')
+        yield ctx.wait_for_external_event(name='event')
+        yield ctx.call_activity(print_activity, input='Finished workflow for version1')
+        return 1
+
+    wfr.start()
+    current_test = 7
+    instance_id = client.schedule_new_workflow(workflow=version1_workflow)
+    time.sleep(3)
+    new_wfr()
+
+    @wfr.versioned_workflow(name="stall_workflow", is_latest=True)
+    def version2_workflow(ctx: wf.DaprWorkflowContext):
+        yield ctx.call_activity(print_activity, input='Received workflow call for version2')
+        yield ctx.wait_for_external_event(name='event')
+        yield ctx.call_activity(print_activity, input='Finished workflow for version2')
+        return 1
+    wfr.start()
+    client.raise_workflow_event(instance_id, event_name='event')
+    time.sleep(2)
+    md = client.get_workflow_state(instance_id)
+    if md.runtime_status == wf.WorkflowStatus.STALLED:
+        print_test('Workflow is stalled')
+    else:
+        print_test('Workflow is not stalled')
+
+def test_patching_stall(client: wf.DaprWorkflowClient):
+    global current_test
+
+    current_test = 8
+    @wfr.workflow
+    def patching_workflow(ctx: wf.DaprWorkflowContext):
+        # This function will be changed throughout the test, to simulate different scenarios
+        return workflow_code(ctx)
+
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        if ctx.is_patched('patch1'):
+            pass
+        else:
+            pass
+        yield ctx.wait_for_external_event(name='event')
+        return 1
+
+    instance_id = client.schedule_new_workflow(workflow=patching_workflow)
+    time.sleep(2)
+
+    def workflow_code(ctx: wf.DaprWorkflowContext):
+        # Removed patch1 check
+        yield ctx.wait_for_external_event(name='event')
+        return 1
+
+    client.raise_workflow_event(instance_id, event_name='event')
+    time.sleep(2)
+    md = client.get_workflow_state(instance_id)
+    if md.runtime_status == wf.WorkflowStatus.STALLED:
+        print_test('Workflow is stalled')
+    else:
+        print_test('Workflow is not stalled')
+
+
+def main():
+    args = sys.argv[1:]
+    if len(args) == 0:
+        print('Usage: python versioning.py <part1|part2>')
+        return
+    if args[0] == 'part1':
+        wfr.start()
+        time.sleep(2)  # wait for workflow runtime to start
+        client = wf.DaprWorkflowClient()
+
+        test_full_versioning(client)
+        test_patching(client)
+
+        test_full_versioning_stall(client)
+        test_patching_stall(client)
+        wfr.shutdown()
+    elif args[0] == 'part2':
+        global current_test
+        current_test = 100
+        print_test('part2')
+        @wfr.versioned_workflow(name="stall_workflow", is_latest=False)
+        def version1_workflow(ctx: wf.DaprWorkflowContext):
+            yield ctx.call_activity(print_activity, input='Received workflow call for version1')
+            yield ctx.wait_for_external_event(name='event')
+            yield ctx.call_activity(print_activity, input='Finished stalled version1 workflow')
+            return 1
+        @wfr.versioned_workflow(name="stall_workflow", is_latest=True)
+        def version2_workflow(ctx: wf.DaprWorkflowContext):
+            yield ctx.call_activity(print_activity, input='Received workflow call for version2')
+            yield ctx.wait_for_external_event(name='event')
+            yield ctx.call_activity(print_activity, input='Finished stalled version2 workflow')
+            return 1
+
+        @wfr.workflow
+        def patching_workflow(ctx: wf.DaprWorkflowContext):
+            if ctx.is_patched('patch1'):
+                pass
+            else:
+                pass
+            yield ctx.wait_for_external_event(name='event')
+            yield ctx.call_activity(print_activity, input='Finished stalled patching workflow')
+            return 1
+
+        wfr.start()
+        time.sleep(10)
+        wfr.shutdown()
+
+if __name__ == '__main__':
+    main()

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py

@@ -154,6 +154,11 @@ def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
         self.__obj.continue_as_new(new_input, save_events=save_events)
 
 
+    def is_patched(self, patch_name: str) -> bool:
+        self._logger.debug(f'{self.instance_id}: Checking if {patch_name} is patched')
+        return self.__obj.is_patched(patch_name)
+
+
 def when_all(tasks: List[task.Task[T]]) -> task.WhenAllTask[T]:
     """Returns a task that completes when all of the provided tasks complete or when one of the
     tasks fail."""