Implements detached workflows from a workflow contexts

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

Author: acroca

dapr/ext/workflow/AGENTS.md

@@ -150,10 +150,22 @@ Passed to workflow functions as the first argument:
 - `instance_id`, `current_utc_datetime`, `is_replaying` — properties
 - `call_activity(activity, *, input, retry_policy, app_id)` → `Task`
 - `call_child_workflow(workflow, *, input, instance_id, retry_policy, app_id)` → `Task`
+- `schedule_new_workflow(workflow, *, input, instance_id, app_id)` → `str` (instance ID; fire-and-forget)
 - `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)`
 
+**Detached vs child workflows** — use `call_child_workflow` when the parent needs to `yield` on the result. Use `schedule_new_workflow` (detached) when the parent should spawn and move on:
+
+- Fire-and-forget: no awaitable Task, returns the spawned instance ID synchronously.
+- No parent linkage on the spawned instance (no completion or failure flows back).
+- Deterministic default instance ID: derived from the parent instance ID + sequence number so replay resolves to the same history record.
+- Cross-app: pass `app_id=...` — the runtime evaluates `WorkflowAccessPolicy` for the target app.
+- Purge/terminate are not recursive across the boundary — the spawned instance manages its own lifecycle.
+- History propagation is intentionally not offered (detached spawns do not inherit the caller's propagated history).
+
+See `examples/workflow/detached.py` for a per-tenant fan-out.
+
 Module-level functions:
 - `when_all(tasks)` → `WhenAllTask` — wait for all tasks to complete
 - `when_any(tasks)` → `WhenAnyTask` — wait for first task to complete

dapr/ext/workflow/_durabletask/internal/helpers.py

@@ -244,6 +244,44 @@ def new_create_child_workflow_action(
     )
 
 
+def new_create_detached_workflow_action(
+    id: int,
+    name: str,
+    instance_id: str,
+    encoded_input: Optional[str],
+    router: Optional[pb.TaskRouter] = None,
+) -> pb.WorkflowAction:
+    """Build a WorkflowAction that spawns a detached workflow.
+
+    The detached workflow is fully decoupled from the caller: no parent
+    linkage is recorded on the new instance and no completion or failure
+    flows back. Detached spawns intentionally do not propagate the caller's
+    history, so no historyPropagationScope is exposed here.
+    """
+    return pb.WorkflowAction(
+        id=id,
+        createDetachedWorkflow=pb.CreateDetachedWorkflowAction(
+            instanceId=instance_id,
+            name=name,
+            input=get_string_value(encoded_input),
+            router=router,
+        ),
+        router=router,
+    )
+
+
+def new_detached_workflow_instance_created_event(
+    event_id: int, instance_id: str
+) -> pb.HistoryEvent:
+    return pb.HistoryEvent(
+        eventId=event_id,
+        timestamp=timestamp_pb2.Timestamp(),
+        detachedWorkflowInstanceCreated=pb.DetachedWorkflowInstanceCreatedEvent(
+            instanceId=instance_id,
+        ),
+    )
+
+
 def is_empty(v: wrappers_pb2.StringValue):
     return v is None or v.value == ''
 

dapr/ext/workflow/_durabletask/task.py

@@ -179,6 +179,46 @@ def call_sub_orchestrator(
         """
         pass
 
+    @abstractmethod
+    def schedule_new_workflow(
+        self,
+        workflow: Union[Orchestrator[TInput, TOutput], str],
+        *,
+        input: Optional[TInput] = None,
+        instance_id: Optional[str] = None,
+        app_id: Optional[str] = None,
+    ) -> str:
+        """Spawn a detached workflow instance and return its ID synchronously.
+
+        Unlike ``call_sub_orchestrator``, the spawned workflow is fully
+        decoupled from the caller: no parent linkage is recorded on the new
+        instance and no completion or failure flows back. There is no
+        awaitable task — the call resolves as soon as the runtime accepts the
+        action.
+
+        Parameters
+        ----------
+        workflow: Orchestrator[TInput, TOutput] | str
+            A reference to the workflow function to spawn, or its registered
+            name (string form is required for cross-app spawns).
+        input: TInput | None
+            Optional JSON-serializable input to pass to the spawned workflow.
+        instance_id: str | None
+            A unique instance ID for the spawned workflow. If not specified,
+            a deterministic ID derived from the caller's instance ID and the
+            current sequence number is used so replay resolves to the same
+            record in history.
+        app_id: str | None
+            The app ID that will execute the spawned workflow. If not
+            specified, the same app as the caller is used.
+
+        Returns
+        -------
+        str
+            The instance ID of the spawned workflow.
+        """
+        pass
+
     @abstractmethod
     def wait_for_external_event(
         self, name: str, *, timeout: Optional[Union[datetime, timedelta]] = None

dapr/ext/workflow/_durabletask/worker.py

@@ -1241,7 +1241,14 @@ def set_complete(
 
         self._is_complete = True
         self._completion_status = status
-        self._pending_actions.clear()  # Cancel any pending actions
+        # Cancel any pending actions except detached-workflow spawns, which are
+        # fire-and-forget: the action is effective the moment schedule_new_workflow
+        # returns, so it must survive the caller's completion.
+        self._pending_actions = {
+            id_: a
+            for id_, a in self._pending_actions.items()
+            if a.HasField('createDetachedWorkflow')
+        }
 
         self._result = result
         result_json: Optional[str] = None
@@ -1465,6 +1472,34 @@ def call_sub_orchestrator(
         )
         return self._pending_tasks.get(id, task.CompletableTask())
 
+    def schedule_new_workflow(
+        self,
+        workflow: Union[task.Orchestrator[TInput, TOutput], str],
+        *,
+        input: Optional[TInput] = None,
+        instance_id: Optional[str] = None,
+        app_id: Optional[str] = None,
+    ) -> str:
+        id = self.next_sequence_number()
+        workflow_name = workflow if isinstance(workflow, str) else task.get_name(workflow)
+        if instance_id is None:
+            instance_id = f'{self.instance_id}:detached:{id:04x}'
+
+        router: Optional[pb.TaskRouter] = None
+        if self._app_id is not None or app_id is not None:
+            router = pb.TaskRouter()
+            if self._app_id is not None:
+                router.sourceAppID = self._app_id
+            if app_id is not None:
+                router.targetAppID = app_id
+
+        encoded_input = shared.to_json(input) if input is not None else None
+        action = ph.new_create_detached_workflow_action(
+            id, workflow_name, instance_id, encoded_input, router
+        )
+        self._pending_actions[id] = action
+        return instance_id
+
     def call_activity_function_helper(
         self,
         id: Optional[int],
@@ -1972,6 +2007,30 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven
                         expected_task_name=event.childWorkflowInstanceCreated.name,
                         actual_task_name=action.createChildWorkflow.name,
                     )
+            elif event.HasField('detachedWorkflowInstanceCreated'):
+                task_id = event.eventId
+                if task_id in ctx._pending_actions and ph.is_optional_timer_action(
+                    ctx._pending_actions[task_id]
+                ):
+                    ctx._drop_optional_pending_at(task_id)
+                action = ctx._pending_actions.pop(task_id, None)
+                if not action:
+                    raise _get_non_determinism_error(
+                        task_id, task.get_name(ctx.schedule_new_workflow)
+                    )
+                elif not action.HasField('createDetachedWorkflow'):
+                    expected_method_name = task.get_name(ctx.schedule_new_workflow)
+                    raise _get_wrong_action_type_error(task_id, expected_method_name, action)
+                elif (
+                    action.createDetachedWorkflow.instanceId
+                    != event.detachedWorkflowInstanceCreated.instanceId
+                ):
+                    raise _get_wrong_action_name_error(
+                        task_id,
+                        method_name=task.get_name(ctx.schedule_new_workflow),
+                        expected_task_name=event.detachedWorkflowInstanceCreated.instanceId,
+                        actual_task_name=action.createDetachedWorkflow.instanceId,
+                    )
             elif event.HasField('childWorkflowInstanceCompleted'):
                 task_id = event.childWorkflowInstanceCompleted.taskScheduledId
                 sub_orch_task = ctx._pending_tasks.pop(task_id, None)
@@ -2239,6 +2298,8 @@ def _get_method_name_for_action(action: pb.WorkflowAction) -> str:
         return task.get_name(task.OrchestrationContext.create_timer)
     elif action_type == 'createChildWorkflow':
         return task.get_name(task.OrchestrationContext.call_sub_orchestrator)
+    elif action_type == 'createDetachedWorkflow':
+        return task.get_name(task.OrchestrationContext.schedule_new_workflow)
     # elif action_type == "sendEvent":
     #    return task.get_name(task.OrchestrationContext.send_event)
     else:

dapr/ext/workflow/dapr_workflow_context.py

@@ -153,6 +153,31 @@ def wf(ctx: task.OrchestrationContext, inp: TInput):
     def get_propagated_history(self) -> Optional[PropagatedHistory]:
         return self.__obj.get_propagated_history()
 
+    def schedule_new_workflow(
+        self,
+        workflow: Union[Workflow, str],
+        *,
+        input: Optional[TInput] = None,
+        instance_id: Optional[str] = None,
+        app_id: Optional[str] = None,
+    ) -> str:
+        if isinstance(workflow, str):
+            workflow_name = workflow
+        elif hasattr(workflow, '_dapr_alternate_name'):
+            workflow_name = workflow.__dict__['_dapr_alternate_name']
+        else:
+            workflow_name = workflow.__name__
+
+        if app_id is not None:
+            self._logger.debug(
+                f'{self.instance_id}: Spawning detached workflow {workflow_name} on app {app_id}'
+            )
+        else:
+            self._logger.debug(f'{self.instance_id}: Spawning detached workflow {workflow_name}')
+        return self.__obj.schedule_new_workflow(
+            workflow_name, input=input, instance_id=instance_id, app_id=app_id
+        )
+
     def wait_for_external_event(
         self,
         name: str,

dapr/ext/workflow/workflow_context.py

@@ -180,6 +180,42 @@ def get_propagated_history(self) -> Optional[PropagatedHistory]:
         no history was propagated."""
         pass
 
+    @abstractmethod
+    def schedule_new_workflow(
+        self,
+        workflow: Union[Workflow[TOutput], str],
+        *,
+        input: Optional[TInput] = None,
+        instance_id: Optional[str] = None,
+        app_id: Optional[str] = None,
+    ) -> str:
+        """Spawn a detached workflow instance and return its ID synchronously.
+
+        Unlike ``call_child_workflow``, the spawned workflow is fully
+        decoupled from the caller: no parent linkage is recorded, no
+        completion or failure flows back, and there is no awaitable task.
+
+        Parameters
+        ----------
+        workflow: Workflow[TOutput] | str
+            A reference to the workflow function, or its registered name
+            (string form is required for cross-app spawns).
+        input: TInput | None
+            Optional JSON-serializable input to pass to the spawned workflow.
+        instance_id: str | None
+            Instance ID for the spawned workflow. When omitted, the runtime
+            derives a deterministic ID from the caller's instance ID so
+            replay resolves to the same history record.
+        app_id: str | None
+            The app ID that will execute the spawned workflow.
+
+        Returns
+        -------
+        str
+            The instance ID of the spawned workflow.
+        """
+        pass
+
     @abstractmethod
     def wait_for_external_event(
         self,

examples/workflow/detached.py

@@ -0,0 +1,78 @@
+# -*- 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.
+
+"""Detached workflow fan-out example.
+
+The parent workflow spawns one detached workflow per tenant. Detached spawns
+are fire-and-forget: the parent receives the spawned instance ID
+synchronously and completes without waiting for the detached workflows to
+finish. Each detached instance runs independently, with no parent linkage
+in its history.
+
+This differs from ``call_child_workflow`` (see child_workflow.py), where the
+parent yields a Task and blocks until the child completes.
+"""
+
+import dapr.ext.workflow as wf
+
+wfr = wf.WorkflowRuntime()
+
+TENANTS = ['acme', 'globex', 'initech']
+
+
+@wfr.workflow
+def parent_workflow(ctx: wf.DaprWorkflowContext, tenants: list[str]):
+    spawned_ids: list[str] = []
+    for tenant in tenants:
+        detached_id = f'tenant-{tenant}'
+        spawned = ctx.schedule_new_workflow(tenant_workflow, input=tenant, instance_id=detached_id)
+        spawned_ids.append(spawned)
+        if not ctx.is_replaying:
+            print(f'*** Spawned detached workflow {spawned}', flush=True)
+    return spawned_ids
+
+
+@wfr.workflow
+def tenant_workflow(ctx: wf.DaprWorkflowContext, tenant: str):
+    if not ctx.is_replaying:
+        print(f'*** Tenant workflow started for {tenant}', flush=True)
+    yield ctx.call_activity(process_tenant, input=tenant)
+    return f'{tenant}-done'
+
+
+@wfr.activity
+def process_tenant(ctx: wf.WorkflowActivityContext, tenant: str) -> str:
+    print(f'*** Processing tenant {tenant}', flush=True)
+    return f'processed:{tenant}'
+
+
+if __name__ == '__main__':
+    wfr.start()
+
+    wf_client = wf.DaprWorkflowClient()
+    parent_id = wf_client.schedule_new_workflow(workflow=parent_workflow, input=TENANTS)
+
+    parent_state = wf_client.wait_for_workflow_completion(parent_id, timeout_in_seconds=30)
+    print(f'*** Parent workflow {parent_id} finished: {parent_state.runtime_status}', flush=True)
+
+    # The detached workflows continue running independently of the parent.
+    # Poll each to confirm they eventually complete.
+    for tenant in TENANTS:
+        detached_id = f'tenant-{tenant}'
+        state = wf_client.wait_for_workflow_completion(detached_id, timeout_in_seconds=30)
+        print(
+            f'*** Detached workflow {detached_id} finished: '
+            f'{state.runtime_status} output={state.serialized_output}',
+            flush=True,
+        )
+
+    wfr.shutdown()