diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 4874a576c..bb7feb964 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -14,6 +14,7 @@ dapr/ext/workflow/ ├── workflow_context.py # WorkflowContext ABC ├── workflow_activity_context.py # WorkflowActivityContext wrapper ├── workflow_state.py # WorkflowState, WorkflowStatus enum +├── workflow_management.py # WorkflowHistoryEvent(Type), WorkflowInstanceIdPage ├── retry_policy.py # RetryPolicy wrapper ├── util.py # gRPC address resolution ├── logger/options.py # LoggerOptions @@ -24,6 +25,7 @@ tests/ext/workflow/ ├── test_dapr_workflow_context.py # Context method proxying ├── test_workflow_activity_context.py # Activity context properties ├── test_workflow_client.py # Sync client (mock gRPC) +├── test_workflow_management.py # list/history/rerun on both clients ├── test_workflow_client_aio.py # Async client (IsolatedAsyncioTestCase) ├── test_workflow_runtime.py # Registration, decorators, worker readiness ├── test_workflow_util.py # Address resolution @@ -86,6 +88,11 @@ from dapr.ext.workflow import ( when_any, # Race combinator — wait for first task alternate_name, # Decorator to set a custom registration name RetryPolicy, # Retry config for activities/child workflows + WorkflowHistoryEvent, # One event from an instance's execution history + WorkflowHistoryEventType, # Enum of history event kinds; unknown ones map to UNKNOWN + WorkflowInstanceIdPage, # One page of instance IDs plus the continuation token + FailureDetails, # Error carried by WorkflowHistoryEvent / TaskFailedError + UNSET, # Sentinel default for rerun_workflow_from_event's input ) # Async client: @@ -139,9 +146,13 @@ Client for workflow lifecycle management: - `terminate_workflow(instance_id, *, output, recursive)` - `pause_workflow(instance_id)` / `resume_workflow(instance_id)` - `purge_workflow(instance_id, *, recursive)` +- `list_workflow_instance_ids(*, page_size, continuation_token)` → `WorkflowInstanceIdPage` +- `iter_workflow_instance_ids(*, page_size=1024)` → iterator over instance IDs, paging internally (`async for` on the async client) +- `get_workflow_history(instance_id)` → `list[WorkflowHistoryEvent]` +- `rerun_workflow_from_event(instance_id, event_id, *, new_instance_id, input, new_child_workflow_instance_id)` → new `instance_id`. Omitting `input` keeps the original; passing `None` clears it. That pair is an `input` + `overwriteInput` pair on the wire, which the engine layer takes as-is; the public method collapses it into one argument via the exported `UNSET` sentinel. - `close()` — close gRPC connection -Converts gRPC "no such instance exists" errors to `None` returns. The async variant in `aio/` has the same API with `async` methods. +`get_workflow_state` converts gRPC "no such instance exists" errors to a `None` return. The other methods let the error propagate, including `get_workflow_history`, which raises NOT_FOUND for a missing or purged instance. The async variant in `aio/` has the same API with `async` methods. ### DaprWorkflowContext (`dapr_workflow_context.py`) diff --git a/dapr/ext/workflow/__init__.py b/dapr/ext/workflow/__init__.py index 228ceac24..7bae529cf 100644 --- a/dapr/ext/workflow/__init__.py +++ b/dapr/ext/workflow/__init__.py @@ -14,7 +14,7 @@ """ # Import your main classes here -from dapr.ext.workflow._durabletask.task import TaskFailedError +from dapr.ext.workflow._durabletask.task import FailureDetails, TaskFailedError from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext, when_all, when_any from dapr.ext.workflow.mcp import DaprMCPClient, MCPToolDef @@ -28,6 +28,12 @@ ) from dapr.ext.workflow.retry_policy import RetryPolicy from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext +from dapr.ext.workflow.workflow_management import ( + UNSET, + WorkflowHistoryEvent, + WorkflowHistoryEventType, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_runtime import WorkflowRuntime, alternate_name from dapr.ext.workflow.workflow_state import WorkflowState, WorkflowStatus @@ -38,11 +44,16 @@ 'WorkflowActivityContext', 'WorkflowState', 'WorkflowStatus', + 'WorkflowHistoryEvent', + 'WorkflowHistoryEventType', + 'WorkflowInstanceIdPage', + 'UNSET', 'when_all', 'when_any', 'alternate_name', 'RetryPolicy', 'TaskFailedError', + 'FailureDetails', 'PropagationScope', 'PropagatedHistory', 'PropagationNotFoundError', diff --git a/dapr/ext/workflow/_durabletask/aio/client.py b/dapr/ext/workflow/_durabletask/aio/client.py index a5da39b14..40d99ee5d 100644 --- a/dapr/ext/workflow/_durabletask/aio/client.py +++ b/dapr/ext/workflow/_durabletask/aio/client.py @@ -38,6 +38,7 @@ TOutput, WorkflowIdReusePolicy, WorkflowState, + _new_rerun_request, _TransientTimeout, new_orchestration_state, ) @@ -307,3 +308,36 @@ async def purge_orchestration(self, instance_id: str, recursive: bool = True): req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive) self._logger.info(f"Purging instance '{instance_id}'.") await self._get_stub().PurgeInstances(req) + + async def list_instance_ids( + self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None + ) -> pb.ListInstanceIDsResponse: + req = pb.ListInstanceIDsRequest(pageSize=page_size, continuationToken=continuation_token) + return await self._get_stub().ListInstanceIDs(req) + + async def get_instance_history(self, instance_id: str) -> list[pb.HistoryEvent]: + req = pb.GetInstanceHistoryRequest(instanceId=instance_id) + res: pb.GetInstanceHistoryResponse = await self._get_stub().GetInstanceHistory(req) + return list(res.events) + + async def rerun_orchestration_from_event( + self, + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str] = None, + input: Optional[Any] = None, + overwrite_input: bool = False, + new_child_instance_id: Optional[str] = None, + ) -> str: + req = _new_rerun_request( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input, + overwrite_input=overwrite_input, + new_child_instance_id=new_child_instance_id, + ) + self._logger.info(f"Rerunning instance '{instance_id}' from event {event_id}.") + res: pb.RerunWorkflowFromEventResponse = await self._get_stub().RerunWorkflowFromEvent(req) + return res.newInstanceID diff --git a/dapr/ext/workflow/_durabletask/client.py b/dapr/ext/workflow/_durabletask/client.py index 07e63d4a7..b629cc94d 100644 --- a/dapr/ext/workflow/_durabletask/client.py +++ b/dapr/ext/workflow/_durabletask/client.py @@ -150,6 +150,46 @@ def new_orchestration_state( ) +def _new_rerun_request( + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str], + input: Optional[Any], + overwrite_input: bool, + new_child_instance_id: Optional[str], +) -> pb.RerunWorkflowFromEventRequest: + """Build a RerunWorkflowFromEvent request. + + ``input`` and ``overwrite_input`` mirror the wire, which needs both: the + replacement input rides in a non-optional ``StringValue``, so "leave the + input alone" and "replace it with null" are only distinguishable through the + flag. Callers of the public client express that as one argument; see + :data:`dapr.ext.workflow.UNSET`. + + Raises: + ValueError: If event_id is negative. The field is uint32 on the wire, so + protobuf would otherwise reject it with a message naming neither the + argument nor the reason. + """ + if event_id < 0: + raise ValueError( + f'event_id must be non-negative, got {event_id}. The runtime reports -1 for ' + 'history events it assigns no ID to, and those cannot be rerun from.' + ) + + return pb.RerunWorkflowFromEventRequest( + sourceInstanceID=instance_id, + eventID=event_id, + newInstanceID=new_instance_id, + input=wrappers_pb2.StringValue(value=shared.to_json(input)) + if overwrite_input and input is not None + else None, + overwriteInput=overwrite_input, + newChildWorkflowInstanceID=new_child_instance_id, + ) + + class TaskHubGrpcClient: def __init__( self, @@ -430,3 +470,36 @@ def purge_orchestration(self, instance_id: str, recursive: bool = True): req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive) self._logger.info(f"Purging instance '{instance_id}'.") self._stub.PurgeInstances(req) + + def list_instance_ids( + self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None + ) -> pb.ListInstanceIDsResponse: + req = pb.ListInstanceIDsRequest(pageSize=page_size, continuationToken=continuation_token) + return self._stub.ListInstanceIDs(req) + + def get_instance_history(self, instance_id: str) -> list[pb.HistoryEvent]: + req = pb.GetInstanceHistoryRequest(instanceId=instance_id) + res: pb.GetInstanceHistoryResponse = self._stub.GetInstanceHistory(req) + return list(res.events) + + def rerun_orchestration_from_event( + self, + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str] = None, + input: Optional[Any] = None, + overwrite_input: bool = False, + new_child_instance_id: Optional[str] = None, + ) -> str: + req = _new_rerun_request( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input, + overwrite_input=overwrite_input, + new_child_instance_id=new_child_instance_id, + ) + self._logger.info(f"Rerunning instance '{instance_id}' from event {event_id}.") + res: pb.RerunWorkflowFromEventResponse = self._stub.RerunWorkflowFromEvent(req) + return res.newInstanceID diff --git a/dapr/ext/workflow/aio/dapr_workflow_client.py b/dapr/ext/workflow/aio/dapr_workflow_client.py index 8a6072cf2..9d56ab945 100644 --- a/dapr/ext/workflow/aio/dapr_workflow_client.py +++ b/dapr/ext/workflow/aio/dapr_workflow_client.py @@ -16,7 +16,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Optional, TypeVar, Union +from typing import Any, AsyncIterator, Optional, TypeVar, Union from warnings import warn from grpc.aio import AioRpcError @@ -31,6 +31,11 @@ from dapr.ext.workflow.logger import Logger, LoggerOptions from dapr.ext.workflow.util import get_grpc_channel_options, getAddress from dapr.ext.workflow.workflow_context import Workflow +from dapr.ext.workflow.workflow_management import ( + UNSET, + WorkflowHistoryEvent, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_state import WorkflowState T = TypeVar('T') @@ -306,3 +311,128 @@ async def purge_workflow(self, instance_id: str, recursive: bool = True) -> None recursive: The optional flag to also purge data from all child workflows. """ return await self.__obj.purge_orchestration(instance_id, recursive) + + async def list_workflow_instance_ids( + self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None + ) -> WorkflowInstanceIdPage: + """Fetches one page of workflow instance IDs for this app. + + The listing is scoped to the app and namespace of the sidecar this + client is connected to. Use iter_workflow_instance_ids instead unless you + need to hold on to the continuation token yourself, for example to + resume paging in a later request. + + Args: + page_size: The maximum number of instance IDs to return. Defaults + to leaving the limit unset, in which case how many come back is up + to the runtime and the state store behind it. + continuation_token: The token from a previous page, to start this + page where that one ended. Defaults to starting from the first page. + + Returns: + A page of instance IDs, and the token for the next page if there is one. + """ + res = await self.__obj.list_instance_ids( + page_size=page_size, continuation_token=continuation_token + ) + return WorkflowInstanceIdPage._from_proto(res) + + async def iter_workflow_instance_ids(self, *, page_size: int = 1024) -> AsyncIterator[str]: + """Iterates over every workflow instance ID for this app, paging as it goes. + + Pages are fetched lazily, so abandoning the iterator early stops the + requests too. Iteration ends on the first page that comes back without + a usable continuation token. + + Args: + page_size: The maximum number of instance IDs to fetch per request. + + Yields: + Instance IDs, in the order the runtime returns them. + """ + continuation_token = None + while True: + page = await self.list_workflow_instance_ids( + page_size=page_size, continuation_token=continuation_token + ) + for instance_id in page.instance_ids: + yield instance_id + if not page.continuation_token: + return + continuation_token = page.continuation_token + + async def get_workflow_history(self, instance_id: str) -> list[WorkflowHistoryEvent]: + """Fetches the full execution history of a workflow instance. + + Args: + instance_id: The unique ID of the workflow instance to read. + + Returns: + The instance's history events, oldest first. + + Raises: + grpc.aio.AioRpcError: With code NOT_FOUND if no such instance exists, + or if it has been purged. + """ + events = await self.__obj.get_instance_history(instance_id) + return [WorkflowHistoryEvent._from_proto(event) for event in events] + + async def rerun_workflow_from_event( + self, + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str] = None, + input: Any = UNSET, + new_child_workflow_instance_id: Optional[str] = None, + ) -> str: + """Starts a new workflow instance that replays a completed one up to an event. + + History up to event_id is replayed rather than re-executed, and + execution resumes from there. The source instance is left untouched. + + The source instance must have reached a terminal state, must not be a + child workflow, and event_id must name an event the runtime can restart + from — a scheduled activity, a created timer, or a created child + workflow. get_workflow_history reports which events qualify via + WorkflowHistoryEvent.is_rerunnable. + + Args: + instance_id: The unique ID of the workflow instance to rerun. + event_id: The WorkflowHistoryEvent.event_id to resume from. This is + the event's own ID, not its position in the history list. + new_instance_id: The ID to give the new instance. Defaults to a + random ID. Pass None rather than an empty string to get the default: + the runtime accepts '' and creates an instance whose ID is empty, + which it then cannot schedule reminders for. + input: Replacement input for the event being rerun. Omit it to keep + the original input; pass None to clear it. Forwarding code that has + to express "not supplied" can pass + :data:`dapr.ext.workflow.UNSET` explicitly. Supplying it at all is + rejected when event_id names a timer, which accepts no input. + new_child_workflow_instance_id: The ID to give the new child + workflow instance. Only accepted when event_id names a child + workflow creation event. + + Returns: + The ID of the new workflow instance. + + Raises: + ValueError: If event_id is negative, which includes the -1 the + runtime reports for history events it assigns no ID to. + grpc.aio.AioRpcError: With code INVALID_ARGUMENT if the source instance + is a child workflow, has not finished, or if event_id names an event + that rejects these arguments — a timer given an input, or a detached + workflow used as the starting point. NOT_FOUND if event_id names any + other event that cannot be rerun from, and ALREADY_EXISTS if + new_instance_id is already in use. + """ + input_supplied = input is not UNSET + return await self.__obj.rerun_orchestration_from_event( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input if input_supplied else None, + overwrite_input=input_supplied, + new_child_instance_id=new_child_workflow_instance_id, + ) diff --git a/dapr/ext/workflow/dapr_workflow_client.py b/dapr/ext/workflow/dapr_workflow_client.py index dca65c200..b03d3dab0 100644 --- a/dapr/ext/workflow/dapr_workflow_client.py +++ b/dapr/ext/workflow/dapr_workflow_client.py @@ -16,7 +16,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Optional, TypeVar, Union +from typing import Any, Iterator, Optional, TypeVar, Union from warnings import warn from grpc import RpcError @@ -30,6 +30,11 @@ from dapr.ext.workflow.logger import Logger, LoggerOptions from dapr.ext.workflow.util import get_grpc_channel_options, getAddress from dapr.ext.workflow.workflow_context import Workflow +from dapr.ext.workflow.workflow_management import ( + UNSET, + WorkflowHistoryEvent, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_state import WorkflowState T = TypeVar('T') @@ -305,6 +310,130 @@ def purge_workflow(self, instance_id: str, recursive: bool = True): """ return self.__obj.purge_orchestration(instance_id, recursive) + def list_workflow_instance_ids( + self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None + ) -> WorkflowInstanceIdPage: + """Fetches one page of workflow instance IDs for this app. + + The listing is scoped to the app and namespace of the sidecar this + client is connected to. Use iter_workflow_instance_ids instead unless you + need to hold on to the continuation token yourself, for example to + resume paging in a later request. + + Args: + page_size: The maximum number of instance IDs to return. Defaults + to leaving the limit unset, in which case how many come back is up + to the runtime and the state store behind it. + continuation_token: The token from a previous page, to start this + page where that one ended. Defaults to starting from the first page. + + Returns: + A page of instance IDs, and the token for the next page if there is one. + """ + res = self.__obj.list_instance_ids( + page_size=page_size, continuation_token=continuation_token + ) + return WorkflowInstanceIdPage._from_proto(res) + + def iter_workflow_instance_ids(self, *, page_size: int = 1024) -> Iterator[str]: + """Iterates over every workflow instance ID for this app, paging as it goes. + + Pages are fetched lazily, so abandoning the iterator early stops the + requests too. Iteration ends on the first page that comes back without + a usable continuation token. + + Args: + page_size: The maximum number of instance IDs to fetch per request. + + Yields: + Instance IDs, in the order the runtime returns them. + """ + continuation_token = None + while True: + page = self.list_workflow_instance_ids( + page_size=page_size, continuation_token=continuation_token + ) + yield from page.instance_ids + if not page.continuation_token: + return + continuation_token = page.continuation_token + + def get_workflow_history(self, instance_id: str) -> list[WorkflowHistoryEvent]: + """Fetches the full execution history of a workflow instance. + + Args: + instance_id: The unique ID of the workflow instance to read. + + Returns: + The instance's history events, oldest first. + + Raises: + grpc.RpcError: With code NOT_FOUND if no such instance exists, or if + it has been purged. + """ + events = self.__obj.get_instance_history(instance_id) + return [WorkflowHistoryEvent._from_proto(event) for event in events] + + def rerun_workflow_from_event( + self, + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str] = None, + input: Any = UNSET, + new_child_workflow_instance_id: Optional[str] = None, + ) -> str: + """Starts a new workflow instance that replays a completed one up to an event. + + History up to event_id is replayed rather than re-executed, and + execution resumes from there. The source instance is left untouched. + + The source instance must have reached a terminal state, must not be a + child workflow, and event_id must name an event the runtime can restart + from — a scheduled activity, a created timer, or a created child + workflow. get_workflow_history reports which events qualify via + WorkflowHistoryEvent.is_rerunnable. + + Args: + instance_id: The unique ID of the workflow instance to rerun. + event_id: The WorkflowHistoryEvent.event_id to resume from. This is + the event's own ID, not its position in the history list. + new_instance_id: The ID to give the new instance. Defaults to a + random ID. Pass None rather than an empty string to get the default: + the runtime accepts '' and creates an instance whose ID is empty, + which it then cannot schedule reminders for. + input: Replacement input for the event being rerun. Omit it to keep + the original input; pass None to clear it. Forwarding code that has + to express "not supplied" can pass + :data:`dapr.ext.workflow.UNSET` explicitly. Supplying it at all is + rejected when event_id names a timer, which accepts no input. + new_child_workflow_instance_id: The ID to give the new child + workflow instance. Only accepted when event_id names a child + workflow creation event. + + Returns: + The ID of the new workflow instance. + + Raises: + ValueError: If event_id is negative, which includes the -1 the + runtime reports for history events it assigns no ID to. + grpc.RpcError: With code INVALID_ARGUMENT if the source instance + is a child workflow, has not finished, or if event_id names an event + that rejects these arguments — a timer given an input, or a detached + workflow used as the starting point. NOT_FOUND if event_id names any + other event that cannot be rerun from, and ALREADY_EXISTS if + new_instance_id is already in use. + """ + input_supplied = input is not UNSET + return self.__obj.rerun_orchestration_from_event( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input if input_supplied else None, + overwrite_input=input_supplied, + new_child_instance_id=new_child_workflow_instance_id, + ) + def close(self): """Closes the gRPC connection used by the client.""" return self.__obj.close() diff --git a/dapr/ext/workflow/workflow_management.py b/dapr/ext/workflow/workflow_management.py new file mode 100644 index 000000000..d2cb8dad5 --- /dev/null +++ b/dapr/ext/workflow/workflow_management.py @@ -0,0 +1,224 @@ +# -*- 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. + +"""Return types for the workflow management APIs. + +These back :meth:`DaprWorkflowClient.list_workflow_instance_ids` and +:meth:`DaprWorkflowClient.get_workflow_history`, on both the sync and the async +client. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from typing import Any, Optional + +import dapr.ext.workflow._durabletask.internal.helpers as pbh +import dapr.ext.workflow._durabletask.internal.protos as pb +from dapr.ext.workflow._durabletask.task import FailureDetails + + +class _Unset: + """Type of :data:`UNSET`. Not instantiated anywhere else.""" + + def __repr__(self) -> str: + return '' + + +UNSET = _Unset() +"""Default for an argument whose absence means something other than ``None``. + +:meth:`DaprWorkflowClient.rerun_workflow_from_event` needs three answers from +one ``input`` argument: leave the original input alone, replace it with a value, +or clear it. ``None`` already means the third, so the default has to be a fourth +thing, and that is this object. + +Pass it explicitly when forwarding an optional input through your own function, +so "the caller gave me nothing" stays distinct from "the caller gave me None":: + + def retry_charge(instance_id, event_id, amount=UNSET): + return client.rerun_workflow_from_event(instance_id, event_id, input=amount) +""" + + +@dataclass(frozen=True) +class WorkflowInstanceIdPage: + """One page of workflow instance IDs. + + Attributes: + instance_ids: The instance IDs in this page, which may be empty. + continuation_token: An opaque cursor marking where this page ended. + Pass it back unchanged as the next call's continuation_token to get + the following page; None means this was the last page. The runtime + forwards it to the state store, so its contents depend on the + component behind the task hub — never parse or construct one, and + do not expect one to survive a component change. Callers using + :meth:`DaprWorkflowClient.iter_workflow_instance_ids` never see it. + """ + + instance_ids: list[str] + continuation_token: Optional[str] + + @classmethod + def _from_proto(cls, res: pb.ListInstanceIDsResponse) -> WorkflowInstanceIdPage: + return cls( + instance_ids=list(res.instanceIds), + continuation_token=res.continuationToken if res.HasField('continuationToken') else None, + ) + + +class WorkflowHistoryEventType(Enum): + """The kind of a workflow history event. + + Any event type the runtime adds but this SDK version does not know maps to + :attr:`UNKNOWN` rather than raising, so a newer sidecar never breaks history + reads. + + Each value is the protobuf field name of the event payload, which is what + makes that fallback possible. Prefer ``.name`` when displaying or persisting + a type: ``.value`` is a wire detail and spelled in camelCase. + """ + + UNKNOWN = 'unknown' + EXECUTION_STARTED = 'executionStarted' + EXECUTION_COMPLETED = 'executionCompleted' + EXECUTION_TERMINATED = 'executionTerminated' + EXECUTION_SUSPENDED = 'executionSuspended' + EXECUTION_RESUMED = 'executionResumed' + EXECUTION_STALLED = 'executionStalled' + TASK_SCHEDULED = 'taskScheduled' + TASK_COMPLETED = 'taskCompleted' + TASK_FAILED = 'taskFailed' + CHILD_WORKFLOW_INSTANCE_CREATED = 'childWorkflowInstanceCreated' + CHILD_WORKFLOW_INSTANCE_COMPLETED = 'childWorkflowInstanceCompleted' + CHILD_WORKFLOW_INSTANCE_FAILED = 'childWorkflowInstanceFailed' + DETACHED_WORKFLOW_INSTANCE_CREATED = 'detachedWorkflowInstanceCreated' + TIMER_CREATED = 'timerCreated' + TIMER_FIRED = 'timerFired' + EVENT_SENT = 'eventSent' + EVENT_RAISED = 'eventRaised' + CONTINUE_AS_NEW = 'continueAsNew' + WORKFLOW_STARTED = 'workflowStarted' + WORKFLOW_COMPLETED = 'workflowCompleted' + + @classmethod + def _missing_(cls, value: object) -> WorkflowHistoryEventType: + return cls.UNKNOWN + + +# The event types this SDK version knows the runtime can restart from. It is a +# snapshot of a server-side rule, so a sidecar on a different version decides for +# itself; anything it rejects comes back as NOT_FOUND or INVALID_ARGUMENT. +_RERUNNABLE_EVENT_TYPES = frozenset( + { + WorkflowHistoryEventType.TASK_SCHEDULED, + WorkflowHistoryEventType.TIMER_CREATED, + WorkflowHistoryEventType.CHILD_WORKFLOW_INSTANCE_CREATED, + } +) + + +def _failure_details_of(payload: Any) -> Optional[FailureDetails]: + """Reads the failure off a history event payload, for the types that carry one. + + Every event type has a payload of a different message type, so asking one of + them for a field only some of the others define is the normal case here, not + an error. An unset payload reads as no failure. + + Args: + payload: The event's payload message, or None for an unset event type. + + Returns: + The failure, or None if this payload has no failure to report. + """ + try: + has_failure = payload.HasField('failureDetails') + except (AttributeError, ValueError): + return None + if not has_failure: + return None + + details = payload.failureDetails + stack_trace = details.stackTrace + return FailureDetails( + details.errorMessage, + details.errorType, + stack_trace.value if not pbh.is_empty(stack_trace) else None, + ) + + +@dataclass(frozen=True) +class WorkflowHistoryEvent: + """A single event from a workflow instance's execution history. + + The payload of a history event depends on its type, so only the fields that + apply to :attr:`event_type` are populated; the rest are None. Use + :attr:`event_id` as the ``event_id`` argument of + :meth:`DaprWorkflowClient.rerun_workflow_from_event`. + + Attributes: + event_id: The event's ID within its instance's history. Not a list + index: rerun matches on this value. The runtime reports -1 for the + events it assigns no ID to; which types those are is the runtime's + choice, so treat -1 as "no ID" rather than inferring the type. + timestamp: When the runtime recorded the event, as a naive UTC + datetime — the same convention as WorkflowState.created_at. + event_type: The kind of event. + name: The name of whatever the event is about, for the event types + that carry one: the activity, child workflow, external event or + timer, and the workflow itself on EXECUTION_STARTED. + task_scheduled_id: For activity and child workflow completion and + failure events, the event_id of the scheduling event they close out. + None means the event type carries no such field at all — TIMER_FIRED + is the notable one, since the wire keeps its correlation elsewhere. + The field has no presence on the wire, so a 0 is equally a real event + ID or one the runtime never set; unlike event_id, there is no + sentinel value to test for. + failure_details: The error, for the failure event types and for an + EXECUTION_COMPLETED that completed a failed workflow. + """ + + event_id: int + timestamp: datetime + event_type: WorkflowHistoryEventType + name: Optional[str] + task_scheduled_id: Optional[int] + failure_details: Optional[FailureDetails] + + @property + def is_rerunnable(self) -> bool: + """Whether this SDK expects rerun_workflow_from_event to accept this event. + + The runtime has the final say, and a sidecar of a different version may + disagree. Treat this as a filter, not a guarantee. + """ + return self.event_type in _RERUNNABLE_EVENT_TYPES + + @classmethod + def _from_proto(cls, event: pb.HistoryEvent) -> WorkflowHistoryEvent: + # Which payload is set is itself the event type, and the payload types + # share field names (`name`, `taskScheduledId`, `failureDetails`) where + # they share meaning, so read them off the payload rather than + # enumerating every event type three times over. + payload_field = event.WhichOneof('eventType') + payload = getattr(event, payload_field) if payload_field else None + + return cls( + event_id=event.eventId, + timestamp=event.timestamp.ToDatetime(), + event_type=WorkflowHistoryEventType(payload_field), + name=getattr(payload, 'name', '') or None, + task_scheduled_id=getattr(payload, 'taskScheduledId', None), + failure_details=_failure_details_of(payload), + ) diff --git a/examples/workflow/README.md b/examples/workflow/README.md index 85a8dabe7..a9ed61c9a 100644 --- a/examples/workflow/README.md +++ b/examples/workflow/README.md @@ -560,6 +560,68 @@ It shows: dapr run --app-id workflow-history-propagation -- python3 history_propagation.py ``` +### Workflow Management (list, history, rerun) + +This example demonstrates the three workflow management APIs on +`DaprWorkflowClient`, using them to recover a failed order without changing the +workflow code: + +- `iter_workflow_instance_ids()` walks every instance ID for this app, paging + behind the scenes. `list_workflow_instance_ids()` returns a single page plus a + continuation token when you want to drive the paging yourself. +- `get_workflow_history()` returns the instance's events as + `WorkflowHistoryEvent` records, carrying the event ID, type, name, the + scheduling event a completion closes out, and the failure details. +- `rerun_workflow_from_event()` starts a *new* instance that replays the + history up to a chosen event, then resumes from there. Passing `input=` + replaces the input of that event; omitting it keeps the original. + +As of runtime 1.18 three event types can be rerun from — a scheduled activity, +a created timer, and a created child workflow. `WorkflowHistoryEvent.is_rerunnable` +carries this SDK's snapshot of that rule so you do not have to hard-code it, but +the sidecar has the final say and one on a different version may disagree. + +Both clients expose all three, `await`-able on +`dapr.ext.workflow.aio.DaprWorkflowClient`, where `iter_workflow_instance_ids()` +is an `async for`. + +```sh +dapr run --app-id workflow-management -- python3 workflow_management.py +``` + +The output should look like this: + +``` +*** validate_order: order of 0 accepted +*** charge_order: refusing to charge 0 +*** first run finished: status=FAILED +*** list: instance present=True +*** history: #-1 WORKFLOW_STARTED name=None +*** history: #-1 EXECUTION_STARTED name=order_workflow +*** history: #1 TASK_SCHEDULED name=validate_order [rerunnable] +*** history: #-1 WORKFLOW_STARTED name=None +*** history: #-1 TASK_COMPLETED name=None closes=#1 +*** history: #2 TASK_SCHEDULED name=charge_order [rerunnable] +*** history: #-1 WORKFLOW_STARTED name=None +*** history: #-1 TASK_FAILED name=None closes=#2 error=amount must be positive, got 0 +*** history: #3 EXECUTION_COMPLETED name=None error=workflow-management-example: Activity task #2 failed: amount must be positive, got 0 +*** rerun started from event #2 (the runtime calls this activity task #2) +*** charge_order: charged 25 +*** rerun finished: status=COMPLETED result="charged 25" +``` + +One entity, two names: the ID you pass to `rerun_workflow_from_event` is the +history event's `event_id`, and the runtime's own error strings call that same +number an activity *task* (`Activity task #2 failed` above). Both spellings come +from durabletask — `RerunWorkflowFromEvent` and `HistoryEvent` on one side, +`taskScheduledId` and the error text on the other. + +`validate_order` prints once across both runs: it had already completed before +the target event, so the rerun replays its recorded result instead of calling it +again. That does not generalise to every activity — one still in flight at the +target event is re-dispatched, so a rerun can execute it a second time. The +runtime reports `event_id` as `-1` for events it assigns no ID to. + ### Async Activities This example fans out several `async def` activities, then aggregates their diff --git a/examples/workflow/workflow_management.py b/examples/workflow/workflow_management.py new file mode 100644 index 000000000..a2801043b --- /dev/null +++ b/examples/workflow/workflow_management.py @@ -0,0 +1,116 @@ +# -*- 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. + +"""Workflow management example: list, history and rerun. + +An order workflow is scheduled with an amount its charge activity rejects, so +the instance fails. The three management APIs then recover it without touching +the workflow code: + +1. ``iter_workflow_instance_ids()`` finds the instance among this app's instance IDs. +2. ``get_workflow_history()`` reads what the instance did and where it stopped. +3. ``rerun_workflow_from_event()`` starts a new instance that replays the + history up to the failed charge, then re-runs that charge with a corrected + input. + +``validate_order`` had already completed before the target event, so the rerun +replays its recorded result rather than calling it again: it prints once across +both instances. That does not generalise to every activity — an activity still +in flight at the target event is re-dispatched. +""" + +import dapr.ext.workflow as wf + +wfr = wf.WorkflowRuntime() + +instance_id = 'workflow-management-example' + + +@wfr.workflow(name='order_workflow') +def order_workflow(ctx: wf.DaprWorkflowContext, amount: int): + yield ctx.call_activity(validate_order, input=amount) + receipt = yield ctx.call_activity(charge_order, input=amount) + return receipt + + +@wfr.activity(name='validate_order') +def validate_order(ctx: wf.WorkflowActivityContext, amount: int) -> str: + print(f'*** validate_order: order of {amount} accepted', flush=True) + return 'valid' + + +@wfr.activity(name='charge_order') +def charge_order(ctx: wf.WorkflowActivityContext, amount: int) -> str: + if amount <= 0: + print(f'*** charge_order: refusing to charge {amount}', flush=True) + raise ValueError(f'amount must be positive, got {amount}') + print(f'*** charge_order: charged {amount}', flush=True) + return f'charged {amount}' + + +def print_history(client: wf.DaprWorkflowClient, workflow_instance_id: str) -> None: + for event in client.get_workflow_history(workflow_instance_id): + rerun_marker = ' [rerunnable]' if event.is_rerunnable else '' + closes = ( + f' closes=#{event.task_scheduled_id}' if event.task_scheduled_id is not None else '' + ) + failure = f' error={event.failure_details.message}' if event.failure_details else '' + print( + f'*** history: #{event.event_id} {event.event_type.name}' + f' name={event.name}{closes}{failure}{rerun_marker}', + flush=True, + ) + + +def find_charge_event_id(client: wf.DaprWorkflowClient, workflow_instance_id: str) -> int: + """Finds the event to rerun from: the scheduling of the failed charge.""" + for event in client.get_workflow_history(workflow_instance_id): + if event.is_rerunnable and event.name == 'charge_order': + return event.event_id + raise RuntimeError('no rerunnable charge_order event in history') + + +def main(): + client = wf.DaprWorkflowClient() + wfr.start() + + client.schedule_new_workflow(order_workflow, input=0, instance_id=instance_id) + state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30) + print(f'*** first run finished: status={state.runtime_status.name}', flush=True) + + listed = list(client.iter_workflow_instance_ids()) + print(f'*** list: instance present={instance_id in listed}', flush=True) + + print_history(client, instance_id) + + charge_event_id = find_charge_event_id(client, instance_id) + rerun_instance_id = client.rerun_workflow_from_event(instance_id, charge_event_id, input=25) + print( + f'*** rerun started from event #{charge_event_id} ' + f'(the runtime calls this activity task #{charge_event_id})', + flush=True, + ) + + rerun_state = client.wait_for_workflow_completion(rerun_instance_id, timeout_in_seconds=30) + print( + f'*** rerun finished: status={rerun_state.runtime_status.name}' + f' result={rerun_state.serialized_output}', + flush=True, + ) + + client.purge_workflow(rerun_instance_id) + client.purge_workflow(instance_id) + wfr.shutdown() + + +if __name__ == '__main__': + main() diff --git a/tests/examples/test_workflow.py b/tests/examples/test_workflow.py index 2d5ec18eb..d559002c0 100644 --- a/tests/examples/test_workflow.py +++ b/tests/examples/test_workflow.py @@ -82,3 +82,30 @@ def test_async_activities(dapr): ) for line in EXPECTED_ASYNC_ACTIVITIES: assert line in output, f'Missing in output: {line}' + + +EXPECTED_WORKFLOW_MANAGEMENT = [ + '*** first run finished: status=FAILED', + '*** list: instance present=True', + '*** history: #1 TASK_SCHEDULED name=validate_order [rerunnable]', + '*** history: #2 TASK_SCHEDULED name=charge_order [rerunnable]', + '*** history: #-1 TASK_FAILED name=None closes=#2 error=amount must be positive, got 0', + '*** rerun started from event #2', + '*** charge_order: charged 25', + '*** rerun finished: status=COMPLETED result="charged 25"', +] + + +@pytest.mark.example_dir('workflow') +def test_workflow_management(dapr): + output = dapr.run( + '--app-id workflow-management -- python3 workflow_management.py', + timeout=90, + ) + for line in EXPECTED_WORKFLOW_MANAGEMENT: + assert line in output, f'Missing in output: {line}' + # The rerun replays the completed validate_order instead of calling it again. + assert output.count('*** validate_order: order of 0 accepted') == 1 + # The [rerunnable] marker renders at end of line, so the TASK_FAILED expectation + # above stays a substring even if that event were wrongly marked rerunnable. + assert 'got 0 [rerunnable]' not in output diff --git a/tests/ext/workflow/durabletask/test_client_management_apis.py b/tests/ext/workflow/durabletask/test_client_management_apis.py new file mode 100644 index 000000000..ccbd09993 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_client_management_apis.py @@ -0,0 +1,254 @@ +# -*- 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. + +"""Engine-client coverage for the workflow management RPCs (list, history, rerun).""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from google.protobuf import timestamp_pb2, wrappers_pb2 + +import dapr.ext.workflow._durabletask.internal.protos as pb +from dapr.ext.workflow._durabletask.aio.client import AsyncTaskHubGrpcClient +from dapr.ext.workflow._durabletask.client import TaskHubGrpcClient + + +def _sync_client() -> TaskHubGrpcClient: + with patch('dapr.ext.workflow._durabletask.internal.shared.get_grpc_channel'): + client = TaskHubGrpcClient() + client._stub = MagicMock() + return client + + +def _async_client() -> AsyncTaskHubGrpcClient: + with patch('dapr.ext.workflow._durabletask.aio.internal.shared.get_grpc_aio_channel'): + client = AsyncTaskHubGrpcClient() + stub = MagicMock() + client._get_stub = lambda: stub + return client + + +def test_list_instance_ids_omits_unset_pagination_fields(): + client = _sync_client() + client._stub.ListInstanceIDs.return_value = pb.ListInstanceIDsResponse() + + client.list_instance_ids() + + req = client._stub.ListInstanceIDs.call_args[0][0] + assert not req.HasField('pageSize') + assert not req.HasField('continuationToken') + + +def test_list_instance_ids_forwards_pagination_fields(): + client = _sync_client() + client._stub.ListInstanceIDs.return_value = pb.ListInstanceIDsResponse() + + client.list_instance_ids(page_size=50, continuation_token='token1') + + req = client._stub.ListInstanceIDs.call_args[0][0] + assert req.pageSize == 50 + assert req.continuationToken == 'token1' + + +def test_list_instance_ids_returns_the_raw_response(): + client = _sync_client() + expected = pb.ListInstanceIDsResponse(instanceIds=['a', 'b'], continuationToken='next') + client._stub.ListInstanceIDs.return_value = expected + + assert client.list_instance_ids() is expected + + +def test_get_instance_history_unwraps_events(): + client = _sync_client() + events = [pb.HistoryEvent(eventId=1), pb.HistoryEvent(eventId=2)] + client._stub.GetInstanceHistory.return_value = pb.GetInstanceHistoryResponse(events=events) + + result = client.get_instance_history('instance1') + + assert [e.eventId for e in result] == [1, 2] + assert client._stub.GetInstanceHistory.call_args[0][0].instanceId == 'instance1' + + +def test_get_instance_history_of_an_empty_history(): + client = _sync_client() + client._stub.GetInstanceHistory.return_value = pb.GetInstanceHistoryResponse() + + assert client.get_instance_history('instance1') == [] + + +def test_rerun_returns_the_new_instance_id(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse( + newInstanceID='rerun1' + ) + + assert client.rerun_orchestration_from_event('instance1', 4) == 'rerun1' + + +def test_rerun_sends_source_instance_and_event_id(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.sourceInstanceID == 'instance1' + assert req.eventID == 4 + + +def test_rerun_without_overwrite_leaves_the_original_input_alone(): + """overwriteInput must stay false, or the runtime nulls the input.""" + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is False + assert not req.HasField('input') + + +def test_rerun_with_overwrite_and_no_input_clears_the_input(): + """The flag without a value is how the wire says "set the input to null".""" + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4, input=None, overwrite_input=True) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is True + assert not req.HasField('input') + + +def test_rerun_with_an_input_serializes_it_to_json(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event( + 'instance1', 4, input={'amount': 10}, overwrite_input=True + ) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is True + assert req.input == wrappers_pb2.StringValue(value='{"amount": 10}') + + +def test_rerun_with_a_falsy_input_still_overwrites(): + """A falsy payload must not be mistaken for an omitted one.""" + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4, input=0, overwrite_input=True) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is True + assert req.input == wrappers_pb2.StringValue(value='0') + + +def test_rerun_rejects_a_negative_event_id_before_calling_the_stub(): + """eventID is uint32 on the wire, so protobuf would reject -1 with a message + naming neither the argument nor the reason. -1 is reachable: it is what the + runtime reports for history events it assigns no ID to.""" + client = _sync_client() + + with pytest.raises(ValueError, match='event_id must be non-negative, got -1'): + client.rerun_orchestration_from_event('instance1', -1) + + client._stub.RerunWorkflowFromEvent.assert_not_called() + + +def test_rerun_accepts_event_id_zero(): + """Zero is a valid event id, so the guard must not key off falsiness.""" + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 0) + + assert client._stub.RerunWorkflowFromEvent.call_args[0][0].eventID == 0 + + +def test_rerun_omits_unset_instance_ids(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert not req.HasField('newInstanceID') + assert not req.HasField('newChildWorkflowInstanceID') + + +def test_rerun_forwards_both_new_instance_ids(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event( + 'instance1', 4, new_instance_id='new1', new_child_instance_id='child1' + ) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.newInstanceID == 'new1' + assert req.newChildWorkflowInstanceID == 'child1' + + +def test_history_timestamps_survive_the_round_trip(): + client = _sync_client() + event = pb.HistoryEvent( + eventId=1, + timestamp=timestamp_pb2.Timestamp(seconds=1700000000), + taskScheduled=pb.TaskScheduledEvent(name='charge'), + ) + client._stub.GetInstanceHistory.return_value = pb.GetInstanceHistoryResponse(events=[event]) + + assert client.get_instance_history('instance1')[0].timestamp.seconds == 1700000000 + + +@pytest.mark.asyncio +async def test_async_list_instance_ids_forwards_pagination_fields(): + client = _async_client() + client._get_stub().ListInstanceIDs = AsyncMock(return_value=pb.ListInstanceIDsResponse()) + + await client.list_instance_ids(page_size=50, continuation_token='token1') + + req = client._get_stub().ListInstanceIDs.call_args[0][0] + assert req.pageSize == 50 + assert req.continuationToken == 'token1' + + +@pytest.mark.asyncio +async def test_async_get_instance_history_unwraps_events(): + client = _async_client() + client._get_stub().GetInstanceHistory = AsyncMock( + return_value=pb.GetInstanceHistoryResponse(events=[pb.HistoryEvent(eventId=3)]) + ) + + result = await client.get_instance_history('instance1') + + assert [e.eventId for e in result] == [3] + + +@pytest.mark.asyncio +async def test_async_rerun_builds_the_same_request_as_the_sync_client(): + client = _async_client() + client._get_stub().RerunWorkflowFromEvent = AsyncMock( + return_value=pb.RerunWorkflowFromEventResponse(newInstanceID='rerun1') + ) + + omitted = await client.rerun_orchestration_from_event('instance1', 4) + req_omitted = client._get_stub().RerunWorkflowFromEvent.call_args[0][0] + + await client.rerun_orchestration_from_event('instance1', 4, input=None, overwrite_input=True) + req_none = client._get_stub().RerunWorkflowFromEvent.call_args[0][0] + + assert omitted == 'rerun1' + assert req_omitted.overwriteInput is False + assert req_none.overwriteInput is True diff --git a/tests/ext/workflow/test_workflow_management.py b/tests/ext/workflow/test_workflow_management.py new file mode 100644 index 000000000..ab2074ab4 --- /dev/null +++ b/tests/ext/workflow/test_workflow_management.py @@ -0,0 +1,555 @@ +# -*- 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 unittest +from unittest import mock + +from google.protobuf import timestamp_pb2, wrappers_pb2 + +import dapr.ext.workflow._durabletask.internal.protos as pb +from dapr.ext.workflow._durabletask.client import _new_rerun_request +from dapr.ext.workflow.aio.dapr_workflow_client import DaprWorkflowClient as AsyncWorkflowClient +from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient +from dapr.ext.workflow.workflow_management import ( + _RERUNNABLE_EVENT_TYPES, + UNSET, + WorkflowHistoryEvent, + WorkflowHistoryEventType, + WorkflowInstanceIdPage, +) + + +def new_history_event(**kwargs) -> pb.HistoryEvent: + kwargs.setdefault('timestamp', timestamp_pb2.Timestamp(seconds=1700000000)) + return pb.HistoryEvent(**kwargs) + + +class FakeTaskHubGrpcClient: + """Stand-in for the engine client, recording what the public client asked for.""" + + def __init__(self): + self.pages = [pb.ListInstanceIDsResponse()] + self.history = [] + self.rerun_result = 'rerun1' + self.list_calls = [] + self.rerun_calls = [] + + def list_instance_ids(self, *, page_size=None, continuation_token=None): + self.list_calls.append((page_size, continuation_token)) + return self.pages[len(self.list_calls) - 1] + + def get_instance_history(self, instance_id: str): + return self.history + + def rerun_orchestration_from_event( + self, + instance_id, + event_id, + *, + new_instance_id=None, + input=None, + overwrite_input=False, + new_child_instance_id=None, + ): + # Build the real request so the fake rejects what the engine would reject. + _new_rerun_request( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input, + overwrite_input=overwrite_input, + new_child_instance_id=new_child_instance_id, + ) + self.rerun_calls.append( + { + 'instance_id': instance_id, + 'event_id': event_id, + 'new_instance_id': new_instance_id, + 'input': input, + 'overwrite_input': overwrite_input, + 'new_child_instance_id': new_child_instance_id, + } + ) + return self.rerun_result + + +class AsyncFakeTaskHubGrpcClient(FakeTaskHubGrpcClient): + async def list_instance_ids(self, *, page_size=None, continuation_token=None): + return super().list_instance_ids(page_size=page_size, continuation_token=continuation_token) + + async def get_instance_history(self, instance_id: str): + return super().get_instance_history(instance_id) + + async def rerun_orchestration_from_event(self, instance_id, event_id, **kwargs): + return super().rerun_orchestration_from_event(instance_id, event_id, **kwargs) + + +def new_client(fake: FakeTaskHubGrpcClient) -> DaprWorkflowClient: + with mock.patch('dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', return_value=fake): + return DaprWorkflowClient() + + +def new_async_client(fake: AsyncFakeTaskHubGrpcClient) -> AsyncWorkflowClient: + with mock.patch( + 'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient', return_value=fake + ): + return AsyncWorkflowClient() + + +class WorkflowHistoryEventTest(unittest.TestCase): + def test_task_scheduled_carries_its_activity_name(self): + event = new_history_event( + eventId=7, taskScheduled=pb.TaskScheduledEvent(name='charge_card') + ) + + result = WorkflowHistoryEvent._from_proto(event) + + self.assertEqual(7, result.event_id) + self.assertEqual(WorkflowHistoryEventType.TASK_SCHEDULED, result.event_type) + self.assertEqual('charge_card', result.name) + self.assertIsNone(result.task_scheduled_id) + self.assertIsNone(result.failure_details) + + def test_completion_events_link_back_to_their_scheduling_event(self): + event = new_history_event(eventId=8, taskCompleted=pb.TaskCompletedEvent(taskScheduledId=7)) + + result = WorkflowHistoryEvent._from_proto(event) + + self.assertEqual(WorkflowHistoryEventType.TASK_COMPLETED, result.event_type) + self.assertEqual(7, result.task_scheduled_id) + self.assertIsNone(result.name) + + def test_failure_events_carry_the_error(self): + failure_details = pb.TaskFailureDetails( + errorMessage='boom', + errorType='ValueError', + stackTrace=wrappers_pb2.StringValue(value='line 1'), + ) + event = new_history_event( + eventId=9, + taskFailed=pb.TaskFailedEvent(taskScheduledId=7, failureDetails=failure_details), + ) + + result = WorkflowHistoryEvent._from_proto(event) + + self.assertEqual('boom', result.failure_details.message) + self.assertEqual('ValueError', result.failure_details.error_type) + self.assertEqual('line 1', result.failure_details.stack_trace) + + def test_failure_without_a_stack_trace_reports_none(self): + event = new_history_event( + taskFailed=pb.TaskFailedEvent( + failureDetails=pb.TaskFailureDetails(errorMessage='boom', errorType='ValueError') + ) + ) + + self.assertIsNone(WorkflowHistoryEvent._from_proto(event).failure_details.stack_trace) + + def test_a_workflow_failure_carries_the_error_too(self): + event = new_history_event( + executionCompleted=pb.ExecutionCompletedEvent( + failureDetails=pb.TaskFailureDetails(errorMessage='boom', errorType='ValueError') + ) + ) + + result = WorkflowHistoryEvent._from_proto(event) + + self.assertEqual(WorkflowHistoryEventType.EXECUTION_COMPLETED, result.event_type) + self.assertEqual('boom', result.failure_details.message) + + def test_a_successful_completion_has_no_error(self): + event = new_history_event( + executionCompleted=pb.ExecutionCompletedEvent( + result=wrappers_pb2.StringValue(value='"done"') + ) + ) + + self.assertIsNone(WorkflowHistoryEvent._from_proto(event).failure_details) + + def test_an_event_type_without_a_name_reports_none(self): + event = new_history_event(executionSuspended=pb.ExecutionSuspendedEvent()) + + self.assertIsNone(WorkflowHistoryEvent._from_proto(event).name) + + def test_an_unset_payload_maps_to_unknown(self): + result = WorkflowHistoryEvent._from_proto(new_history_event(eventId=1)) + + self.assertEqual(WorkflowHistoryEventType.UNKNOWN, result.event_type) + self.assertFalse(result.is_rerunnable) + + def test_an_unrecognised_event_type_maps_to_unknown_rather_than_raising(self): + """A newer sidecar must not break history reads.""" + self.assertEqual( + WorkflowHistoryEventType.UNKNOWN, + WorkflowHistoryEventType('somethingTheRuntimeAddedLater'), + ) + + def test_every_event_type_the_proto_defines_is_mapped(self): + oneof_fields = { + field.name for field in pb.HistoryEvent.DESCRIPTOR.oneofs_by_name['eventType'].fields + } + mapped = {member.value for member in WorkflowHistoryEventType} - { + WorkflowHistoryEventType.UNKNOWN.value + } + + self.assertEqual(set(), oneof_fields - mapped) + + def test_only_the_three_restartable_event_types_are_rerunnable(self): + rerunnable = { + event_type + for event_type in WorkflowHistoryEventType + if WorkflowHistoryEvent( + event_id=1, + timestamp=None, + event_type=event_type, + name=None, + task_scheduled_id=None, + failure_details=None, + ).is_rerunnable + } + + self.assertEqual(_RERUNNABLE_EVENT_TYPES, rerunnable) + self.assertEqual( + { + WorkflowHistoryEventType.TASK_SCHEDULED, + WorkflowHistoryEventType.TIMER_CREATED, + WorkflowHistoryEventType.CHILD_WORKFLOW_INSTANCE_CREATED, + }, + rerunnable, + ) + + +class WorkflowInstanceIdPageTest(unittest.TestCase): + def test_a_middle_page_carries_the_next_token(self): + res = pb.ListInstanceIDsResponse(instanceIds=['a', 'b'], continuationToken='next') + + page = WorkflowInstanceIdPage._from_proto(res) + + self.assertEqual(['a', 'b'], page.instance_ids) + self.assertEqual('next', page.continuation_token) + + def test_the_last_page_reports_no_token(self): + page = WorkflowInstanceIdPage._from_proto(pb.ListInstanceIDsResponse(instanceIds=['a'])) + + self.assertIsNone(page.continuation_token) + + def test_an_explicitly_empty_token_is_still_a_token(self): + """HasField, not truthiness, decides: an empty token is set, not absent.""" + res = pb.ListInstanceIDsResponse(instanceIds=[], continuationToken='') + + self.assertEqual('', WorkflowInstanceIdPage._from_proto(res).continuation_token) + + +class ListWorkflowInstanceIdsTest(unittest.TestCase): + def test_passes_pagination_arguments_through(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.list_workflow_instance_ids(page_size=25, continuation_token='token1') + + self.assertEqual([(25, 'token1')], fake.list_calls) + + def test_defaults_to_no_pagination_arguments(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.list_workflow_instance_ids() + + self.assertEqual([(None, None)], fake.list_calls) + + def test_returns_the_converted_page(self): + fake = FakeTaskHubGrpcClient() + fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='next')] + client = new_client(fake) + + page = client.list_workflow_instance_ids() + + self.assertEqual( + WorkflowInstanceIdPage(instance_ids=['a'], continuation_token='next'), page + ) + + +class IterWorkflowInstanceIdsTest(unittest.TestCase): + def test_follows_the_continuation_token_across_pages(self): + fake = FakeTaskHubGrpcClient() + fake.pages = [ + pb.ListInstanceIDsResponse(instanceIds=['a', 'b'], continuationToken='page2'), + pb.ListInstanceIDsResponse(instanceIds=['c'], continuationToken='page3'), + pb.ListInstanceIDsResponse(instanceIds=['d']), + ] + client = new_client(fake) + + self.assertEqual(['a', 'b', 'c', 'd'], list(client.iter_workflow_instance_ids())) + self.assertEqual( + [(1024, None), (1024, 'page2'), (1024, 'page3')], + fake.list_calls, + ) + + def test_stops_on_the_first_page_when_there_is_no_token(self): + fake = FakeTaskHubGrpcClient() + fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'])] + client = new_client(fake) + + self.assertEqual(['a'], list(client.iter_workflow_instance_ids())) + self.assertEqual(1, len(fake.list_calls)) + + def test_yields_nothing_when_the_app_has_no_instances(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + self.assertEqual([], list(client.iter_workflow_instance_ids())) + + def test_keeps_paging_through_an_empty_page_that_carries_a_token(self): + fake = FakeTaskHubGrpcClient() + fake.pages = [ + pb.ListInstanceIDsResponse(instanceIds=[], continuationToken='page2'), + pb.ListInstanceIDsResponse(instanceIds=['a']), + ] + client = new_client(fake) + + self.assertEqual(['a'], list(client.iter_workflow_instance_ids())) + + def test_stops_on_an_empty_token_instead_of_looping_forever(self): + """An empty token is not a usable cursor: stores that emit one read it as + "first page", so following it would re-yield page 1 indefinitely.""" + fake = FakeTaskHubGrpcClient() + fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='')] + client = new_client(fake) + + self.assertEqual(['a'], list(client.iter_workflow_instance_ids())) + self.assertEqual(1, len(fake.list_calls)) + + def test_fetches_lazily(self): + fake = FakeTaskHubGrpcClient() + fake.pages = [ + pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='page2'), + pb.ListInstanceIDsResponse(instanceIds=['b']), + ] + client = new_client(fake) + + instances = client.iter_workflow_instance_ids() + next(instances) + + self.assertEqual(1, len(fake.list_calls)) + + def test_honours_the_page_size(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + list(client.iter_workflow_instance_ids(page_size=10)) + + self.assertEqual([(10, None)], fake.list_calls) + + +class GetWorkflowHistoryTest(unittest.TestCase): + def test_converts_every_event(self): + fake = FakeTaskHubGrpcClient() + fake.history = [ + new_history_event(eventId=1, executionStarted=pb.ExecutionStartedEvent(name='order')), + new_history_event(eventId=2, taskScheduled=pb.TaskScheduledEvent(name='charge')), + ] + client = new_client(fake) + + history = client.get_workflow_history('instance1') + + self.assertEqual( + [ + (1, WorkflowHistoryEventType.EXECUTION_STARTED, 'order'), + (2, WorkflowHistoryEventType.TASK_SCHEDULED, 'charge'), + ], + [(e.event_id, e.event_type, e.name) for e in history], + ) + + def test_an_empty_history_is_an_empty_list(self): + client = new_client(FakeTaskHubGrpcClient()) + + self.assertEqual([], client.get_workflow_history('instance1')) + + +class RerunWorkflowFromEventTest(unittest.TestCase): + def test_returns_the_new_instance_id(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + self.assertEqual('rerun1', client.rerun_workflow_from_event('instance1', 4)) + + def test_omitting_input_does_not_ask_the_engine_to_overwrite(self): + """Otherwise the runtime would clear an input the caller never mentioned.""" + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event('instance1', 4) + + self.assertFalse(fake.rerun_calls[0]['overwrite_input']) + + def test_passing_the_sentinel_explicitly_matches_omitting_it(self): + """Forwarding code needs UNSET to mean exactly "not supplied".""" + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event('instance1', 4, input=UNSET) + + self.assertFalse(fake.rerun_calls[0]['overwrite_input']) + + def test_the_sentinel_is_exported_for_forwarding_code(self): + import dapr.ext.workflow as wf + + self.assertIs(UNSET, wf.UNSET) + self.assertEqual('', repr(wf.UNSET)) + + def test_an_explicit_none_input_asks_the_engine_to_overwrite(self): + """None is a value, not an omission: it clears the input.""" + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event('instance1', 4, input=None) + + self.assertIsNone(fake.rerun_calls[0]['input']) + self.assertTrue(fake.rerun_calls[0]['overwrite_input']) + + def test_rejects_a_negative_event_id(self): + """WorkflowHistoryEvent.event_id is -1 for events the runtime gives no ID, + so passing one straight back is a reachable mistake.""" + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + with self.assertRaises(ValueError) as caught: + client.rerun_workflow_from_event('instance1', -1) + + self.assertIn('event_id must be non-negative', str(caught.exception)) + self.assertEqual([], fake.rerun_calls) + + def test_forwards_every_argument(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event( + 'instance1', + 4, + new_instance_id='new1', + input={'amount': 10}, + new_child_workflow_instance_id='child1', + ) + + self.assertEqual( + { + 'instance_id': 'instance1', + 'event_id': 4, + 'new_instance_id': 'new1', + 'input': {'amount': 10}, + 'overwrite_input': True, + 'new_child_instance_id': 'child1', + }, + fake.rerun_calls[0], + ) + + +class AsyncWorkflowManagementTest(unittest.IsolatedAsyncioTestCase): + async def test_list_returns_the_converted_page(self): + fake = AsyncFakeTaskHubGrpcClient() + fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='next')] + client = new_async_client(fake) + + page = await client.list_workflow_instance_ids(page_size=25) + + self.assertEqual( + WorkflowInstanceIdPage(instance_ids=['a'], continuation_token='next'), page + ) + self.assertEqual([(25, None)], fake.list_calls) + + async def test_iter_follows_the_continuation_token_across_pages(self): + fake = AsyncFakeTaskHubGrpcClient() + fake.pages = [ + pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='page2'), + pb.ListInstanceIDsResponse(instanceIds=['b']), + ] + client = new_async_client(fake) + + self.assertEqual( + ['a', 'b'], [instance_id async for instance_id in client.iter_workflow_instance_ids()] + ) + self.assertEqual([(1024, None), (1024, 'page2')], fake.list_calls) + + async def test_iter_stops_on_an_empty_token_instead_of_looping_forever(self): + fake = AsyncFakeTaskHubGrpcClient() + fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='')] + client = new_async_client(fake) + + collected = [instance_id async for instance_id in client.iter_workflow_instance_ids()] + + self.assertEqual(['a'], collected) + self.assertEqual(1, len(fake.list_calls)) + + async def test_iter_fetches_lazily(self): + """An async generator body does not start until the first __anext__, so + laziness here is a different mechanism from the sync generator's.""" + fake = AsyncFakeTaskHubGrpcClient() + fake.pages = [ + pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='page2'), + pb.ListInstanceIDsResponse(instanceIds=['b']), + ] + client = new_async_client(fake) + + instances = client.iter_workflow_instance_ids() + self.assertEqual([], fake.list_calls) + + self.assertEqual('a', await anext(instances)) + self.assertEqual(1, len(fake.list_calls)) + + async def test_get_history_converts_every_event(self): + fake = AsyncFakeTaskHubGrpcClient() + fake.history = [new_history_event(eventId=2, taskScheduled=pb.TaskScheduledEvent(name='c'))] + client = new_async_client(fake) + + history = await client.get_workflow_history('instance1') + + self.assertEqual([(2, 'c')], [(e.event_id, e.name) for e in history]) + + async def test_rerun_omitting_input_does_not_ask_the_engine_to_overwrite(self): + fake = AsyncFakeTaskHubGrpcClient() + client = new_async_client(fake) + + result = await client.rerun_workflow_from_event('instance1', 4) + + self.assertEqual('rerun1', result) + self.assertFalse(fake.rerun_calls[0]['overwrite_input']) + + async def test_rerun_forwards_every_argument(self): + fake = AsyncFakeTaskHubGrpcClient() + client = new_async_client(fake) + + await client.rerun_workflow_from_event( + 'instance1', + 4, + new_instance_id='new1', + input=None, + new_child_workflow_instance_id='child1', + ) + + self.assertEqual( + { + 'instance_id': 'instance1', + 'event_id': 4, + 'new_instance_id': 'new1', + 'input': None, + 'overwrite_input': True, + 'new_child_instance_id': 'child1', + }, + fake.rerun_calls[0], + ) + + +if __name__ == '__main__': + unittest.main()