Add ctx.schedule_new_workflow for detached workflows

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, start_at, app_id, app_namespace)` → `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,48 @@ 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,
+    scheduled_start_timestamp: Optional[timestamp_pb2.Timestamp] = 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.
+    """
+    detached = pb.CreateDetachedWorkflowAction(
+        instanceId=instance_id,
+        name=name,
+        input=get_string_value(encoded_input),
+        router=router,
+    )
+    if scheduled_start_timestamp is not None:
+        detached.scheduledStartTimestamp.CopyFrom(scheduled_start_timestamp)
+    return pb.WorkflowAction(
+        id=id,
+        createDetachedWorkflow=detached,
+        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,55 @@ 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,
+        start_at: Optional[datetime] = None,
+        app_id: Optional[str] = None,
+        app_namespace: 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 a
+            dedicated detached counter is used so replay resolves to the
+            same record in history.
+        start_at: datetime | None
+            Wall-clock time at which the spawned workflow should begin
+            executing. If not specified or in the past, the workflow starts
+            immediately.
+        app_id: str | None
+            The app ID that will execute the spawned workflow. If not
+            specified, the same app as the caller is used.
+        app_namespace: str | None
+            The Dapr namespace of the target app. Only meaningful with
+            ``app_id``.
+
+        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

@@ -1170,6 +1170,7 @@ def __init__(self, instance_id: str):
         self._pending_actions: dict[int, pb.WorkflowAction] = {}
         self._pending_tasks: dict[int, task.CompletableTask] = {}
         self._sequence_number = 0
+        self._detached_counter = 0
         self._current_utc_datetime = datetime(1000, 1, 1)
         self._instance_id = instance_id
         self._app_id = None
@@ -1241,7 +1242,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 +1473,49 @@ 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,
+        start_at: Optional[datetime] = None,
+        app_id: Optional[str] = None,
+        app_namespace: 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:
+            self._detached_counter += 1
+            instance_id = f'{self.instance_id}-{self._detached_counter}'
+
+        router: Optional[pb.TaskRouter] = None
+        if self._app_id is not None or app_id is not None or app_namespace 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
+            if app_namespace is not None:
+                router.targetAppNamespace = app_namespace
+
+        scheduled_start: Optional[timestamp_pb2.Timestamp] = None
+        if start_at is not None:
+            scheduled_start = timestamp_pb2.Timestamp()
+            scheduled_start.FromDatetime(start_at)
+
+        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,
+            scheduled_start_timestamp=scheduled_start,
+        )
+        self._pending_actions[id] = action
+        return instance_id
+
     def call_activity_function_helper(
         self,
         id: Optional[int],
@@ -1972,6 +2023,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 +2314,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,38 @@ 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,
+        start_at: Optional[datetime] = None,
+        app_id: Optional[str] = None,
+        app_namespace: 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,
+            start_at=start_at,
+            app_id=app_id,
+            app_namespace=app_namespace,
+        )
+
     def wait_for_external_event(
         self,
         name: str,

dapr/ext/workflow/workflow_context.py

@@ -180,6 +180,51 @@ 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,
+        start_at: Optional[datetime] = None,
+        app_id: Optional[str] = None,
+        app_namespace: 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.
+        start_at: datetime | None
+            Wall-clock time at which the spawned workflow should begin
+            executing. If not specified or in the past, the workflow starts
+            immediately.
+        app_id: str | None
+            The app ID that will execute the spawned workflow.
+        app_namespace: str | None
+            The Dapr namespace of the target app. Only meaningful with
+            ``app_id``.
+
+        Returns
+        -------
+        str
+            The instance ID of the spawned workflow.
+        """
+        pass
+
     @abstractmethod
     def wait_for_external_event(
         self,