From 03a129398439de14aead000a1c0a3f132d38d9fd Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Thu, 17 Sep 2026 13:46:49 +0200 Subject: [PATCH 1/3] Add workflow management APIs: list, history and rerun The three advanced workflow management operations from dapr/dapr#9729 had no Python surface: the vendored durabletask protos carried ListInstanceIDs, GetInstanceHistory and RerunWorkflowFromEvent, but neither client layer exposed them, so reaching them meant using the gRPC stub directly. DaprWorkflowClient and its async counterpart now expose: - list_workflow_instances(page_size, continuation_token) -> one page of instance IDs plus the token for the next, when the caller wants to hold the cursor themselves. - iter_workflow_instances(page_size) -> a lazy iterator that pages internally; an async generator on the async client. - get_workflow_history(instance_id) -> the instance's events as WorkflowHistoryEvent records. - rerun_workflow_from_event(instance_id, event_id, ...) -> the ID of a new instance that replays history up to the chosen event and resumes there. The rerun input is a single argument rather than a value plus a flag. The wire format pairs a non-optional StringValue with an overwriteInput bool precisely because StringValue cannot express absence, so the two are collapsed behind a sentinel default: omitting input keeps the original, passing None clears it. A falsy value such as 0 still overwrites. WorkflowHistoryEvent carries event_id, timestamp, event_type, name, task_scheduled_id and failure_details, which is enough to choose a rerun point by activity name instead of by raw event number. is_rerunnable reports this SDK's snapshot of which event types the runtime restarts from; the sidecar keeps the final say. Unrecognised event types map to UNKNOWN rather than raising, so a newer sidecar cannot break history reads. Verified end to end against runtime 1.18.0: a failed order is listed, its history read, and the failed charge rerun with a corrected input to completion. examples/workflow/workflow_management.py covers that flow and is asserted by tests/examples/test_workflow.py. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Javier Aliaga --- dapr/ext/workflow/AGENTS.md | 6 + dapr/ext/workflow/__init__.py | 11 +- dapr/ext/workflow/_durabletask/aio/client.py | 34 ++ dapr/ext/workflow/_durabletask/client.py | 73 +++ dapr/ext/workflow/aio/dapr_workflow_client.py | 125 ++++- dapr/ext/workflow/dapr_workflow_client.py | 124 ++++- dapr/ext/workflow/workflow_management.py | 194 +++++++ examples/workflow/README.md | 56 ++ examples/workflow/workflow_management.py | 115 ++++ tests/examples/test_workflow.py | 27 + .../test_client_management_apis.py | 240 +++++++++ .../ext/workflow/test_workflow_management.py | 496 ++++++++++++++++++ 12 files changed, 1498 insertions(+), 3 deletions(-) create mode 100644 dapr/ext/workflow/workflow_management.py create mode 100644 examples/workflow/workflow_management.py create mode 100644 tests/ext/workflow/durabletask/test_client_management_apis.py create mode 100644 tests/ext/workflow/test_workflow_management.py diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 4874a576c..399634284 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 @@ -139,6 +141,10 @@ 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_instances(*, page_size, continuation_token)` → `WorkflowInstanceIdPage` +- `iter_workflow_instances(*, 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 one `overwriteInput` flag on the wire; the sentinel default (`client.UNSET`) is what keeps it a single argument. - `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. diff --git a/dapr/ext/workflow/__init__.py b/dapr/ext/workflow/__init__.py index 228ceac24..626f5ec05 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,11 @@ ) from dapr.ext.workflow.retry_policy import RetryPolicy from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext +from dapr.ext.workflow.workflow_management import ( + WorkflowHistoryEvent, + WorkflowHistoryEventType, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_runtime import WorkflowRuntime, alternate_name from dapr.ext.workflow.workflow_state import WorkflowState, WorkflowStatus @@ -38,11 +43,15 @@ 'WorkflowActivityContext', 'WorkflowState', 'WorkflowStatus', + 'WorkflowHistoryEvent', + 'WorkflowHistoryEventType', + 'WorkflowInstanceIdPage', '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..d9ef654bb 100644 --- a/dapr/ext/workflow/_durabletask/aio/client.py +++ b/dapr/ext/workflow/_durabletask/aio/client.py @@ -32,13 +32,16 @@ get_grpc_aio_channel, ) from dapr.ext.workflow._durabletask.client import ( + UNSET, OrchestrationStatus, TaskHubGrpcClient, TInput, TOutput, WorkflowIdReusePolicy, WorkflowState, + _new_rerun_request, _TransientTimeout, + _Unset, new_orchestration_state, ) @@ -307,3 +310,34 @@ 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: Union[Any, _Unset] = UNSET, + new_child_instance_id: Optional[str] = None, + ) -> str: + req = _new_rerun_request( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=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..e70000d8a 100644 --- a/dapr/ext/workflow/_durabletask/client.py +++ b/dapr/ext/workflow/_durabletask/client.py @@ -33,6 +33,20 @@ class _TransientTimeout(Exception): budget. Callers convert this to a public ``TimeoutError``.""" +class _Unset: + """Sentinel for an argument that was not supplied. + + Distinct from ``None``, which is a meaningful value for the rerun input: + omitting ``input`` keeps the original, passing ``None`` clears it. + """ + + def __repr__(self) -> str: + return '' + + +UNSET = _Unset() + + TInput = TypeVar('TInput') TOutput = TypeVar('TOutput') @@ -150,6 +164,34 @@ def new_orchestration_state( ) +def _new_rerun_request( + instance_id: str, + event_id: int, + *, + new_instance_id: Optional[str], + input: Union[Any, _Unset], + new_child_instance_id: Optional[str], +) -> pb.RerunWorkflowFromEventRequest: + """Build a RerunWorkflowFromEvent request, resolving the input sentinel. + + The wire format carries the replacement input in a non-optional + ``StringValue`` plus an ``overwriteInput`` flag, so "leave the input alone" + and "replace it with null" are only distinguishable through the flag. The + sentinel collapses that pair into a single argument. + """ + overwrite_input = not isinstance(input, _Unset) + 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 +472,34 @@ 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: Union[Any, _Unset] = UNSET, + new_child_instance_id: Optional[str] = None, + ) -> str: + req = _new_rerun_request( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=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..c1bc09fdd 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,10 @@ 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 ( + WorkflowHistoryEvent, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_state import WorkflowState T = TypeVar('T') @@ -306,3 +310,122 @@ 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_instances( + 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_instances 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_instances(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_instances( + 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 = client.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. 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: + 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. + """ + return await self.__obj.rerun_orchestration_from_event( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input, + 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..60174da68 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,10 @@ 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 ( + WorkflowHistoryEvent, + WorkflowInstanceIdPage, +) from dapr.ext.workflow.workflow_state import WorkflowState T = TypeVar('T') @@ -305,6 +309,124 @@ def purge_workflow(self, instance_id: str, recursive: bool = True): """ return self.__obj.purge_orchestration(instance_id, recursive) + def list_workflow_instances( + 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_instances 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_instances(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_instances( + 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 = client.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. 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: + 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. + """ + return self.__obj.rerun_orchestration_from_event( + instance_id, + event_id, + new_instance_id=new_instance_id, + input=input, + 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..1dee70d22 --- /dev/null +++ b/dapr/ext/workflow/workflow_management.py @@ -0,0 +1,194 @@ +# -*- 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_instances` 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 + + +@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: Token to pass to the next + :meth:`DaprWorkflowClient.list_workflow_instances` call, or None + when this is the last page. + """ + + 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. + TIMER_FIRED does not carry it; the wire puts that correlation in a + different field this type does not surface. + 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..046c56178 100644 --- a/examples/workflow/README.md +++ b/examples/workflow/README.md @@ -560,6 +560,62 @@ 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_instances()` walks every instance ID for this app, paging + behind the scenes. `list_workflow_instances()` 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_instances()` +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 +*** charge_order: charged 25 +*** rerun finished: status=COMPLETED result="charged 25" +``` + +`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..50901eb81 --- /dev/null +++ b/examples/workflow/workflow_management.py @@ -0,0 +1,115 @@ +# -*- 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_instances()`` finds the instance among this app's instances. +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. +""" + +from time import sleep + +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() + sleep(5) + + 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_instances()) + 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}', 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..14c3b3a74 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_client_management_apis.py @@ -0,0 +1,240 @@ +# -*- 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 UNSET, 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_input_leaves_the_original_input_alone(): + """Omitting input must not set overwriteInput, 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_none_input_clears_the_input(): + """None is a value, not an omission: it overwrites the input with null.""" + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4, input=None) + + 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}) + + 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) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is True + assert req.input == wrappers_pb2.StringValue(value='0') + + +def test_rerun_explicit_unset_matches_omitting_input(): + client = _sync_client() + client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() + + client.rerun_orchestration_from_event('instance1', 4, input=UNSET) + + req = client._stub.RerunWorkflowFromEvent.call_args[0][0] + assert req.overwriteInput is False + + +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_resolves_the_input_sentinel_like_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) + 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..12be61127 --- /dev/null +++ b/tests/ext/workflow/test_workflow_management.py @@ -0,0 +1,496 @@ +# -*- 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 UNSET +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, + 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=UNSET, + new_child_instance_id=None, + ): + self.rerun_calls.append( + { + 'instance_id': instance_id, + 'event_id': event_id, + 'new_instance_id': new_instance_id, + 'input': 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 ListWorkflowInstancesTest(unittest.TestCase): + def test_passes_pagination_arguments_through(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.list_workflow_instances(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_instances() + + 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_instances() + + self.assertEqual( + WorkflowInstanceIdPage(instance_ids=['a'], continuation_token='next'), page + ) + + +class IterWorkflowInstancesTest(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_instances())) + 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_instances())) + 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_instances())) + + 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_instances())) + + 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_instances())) + 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_instances() + 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_instances(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_reaches_the_engine_as_the_sentinel(self): + """None must not leak in as a default, or the runtime clears the input.""" + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event('instance1', 4) + + self.assertIs(UNSET, fake.rerun_calls[0]['input']) + + def test_an_explicit_none_input_reaches_the_engine_as_none(self): + fake = FakeTaskHubGrpcClient() + client = new_client(fake) + + client.rerun_workflow_from_event('instance1', 4, input=None) + + self.assertIsNone(fake.rerun_calls[0]['input']) + + 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}, + '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_instances(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_instances()] + ) + 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_instances()] + + self.assertEqual(['a'], collected) + 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_reaches_the_engine_as_the_sentinel(self): + fake = AsyncFakeTaskHubGrpcClient() + client = new_async_client(fake) + + result = await client.rerun_workflow_from_event('instance1', 4) + + self.assertEqual('rerun1', result) + self.assertIs(UNSET, fake.rerun_calls[0]['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, + 'new_child_instance_id': 'child1', + }, + fake.rerun_calls[0], + ) + + +if __name__ == '__main__': + unittest.main() From 16e5a27a2f026cdbbee2f738d641fb2f5922af6c Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Fri, 18 Sep 2026 09:47:31 +0200 Subject: [PATCH 2/3] Narrow the AGENTS.md note on NOT_FOUND handling The blanket statement that the client converts "no such instance exists" to a None return only ever described get_workflow_state; the other methods propagate. Adding get_workflow_history, which raises NOT_FOUND for a missing or purged instance, made the sentence actively misleading. Also covers the input sentinel's repr, which is what help() and tracebacks show for the rerun default. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Javier Aliaga --- dapr/ext/workflow/AGENTS.md | 2 +- .../ext/workflow/durabletask/test_client_management_apis.py | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 399634284..8fbe628ba 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -147,7 +147,7 @@ Client for workflow lifecycle management: - `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 one `overwriteInput` flag on the wire; the sentinel default (`client.UNSET`) is what keeps it a single argument. - `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/tests/ext/workflow/durabletask/test_client_management_apis.py b/tests/ext/workflow/durabletask/test_client_management_apis.py index 14c3b3a74..5ae08a54f 100644 --- a/tests/ext/workflow/durabletask/test_client_management_apis.py +++ b/tests/ext/workflow/durabletask/test_client_management_apis.py @@ -37,6 +37,12 @@ def _async_client() -> AsyncTaskHubGrpcClient: return client +def test_the_input_sentinel_reads_as_unset(): + """UNSET is the documented default of a public argument, so its repr shows up + in help() output, IDE hovers and tracebacks.""" + assert repr(UNSET) == '' + + def test_list_instance_ids_omits_unset_pagination_fields(): client = _sync_client() client._stub.ListInstanceIDs.return_value = pb.ListInstanceIDsResponse() From cb2af6628917115d7ca4fa6047d781bcf93bab06 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Tue, 22 Sep 2026 17:40:15 +0200 Subject: [PATCH 3/3] Address review feedback on the workflow management APIs Renames the listing methods to say what they return. They hand back instance IDs, not workflows, and a reviewer read them the other way. list_workflow_instance_ids and iter_workflow_instance_ids also line up with java-sdk#1798's listInstanceIds, and leave the unqualified name free if a filtered list returning WorkflowState ever lands. Moves the rerun input sentinel to the public API surface. The wire needs an input field plus an overwriteInput flag, so the engine layer now takes exactly that pair and knows nothing about sentinels. UNSET is defined in workflow_management.py and exported, which is what forwarding code needs: a wrapper passing an optional input through could not previously say "not supplied" without importing from a private module. Rejects a negative event_id before building the request. eventID is uint32 on the wire, so protobuf refused it with a message naming neither the argument nor the reason, and -1 is reachable precisely because it is what the runtime reports for history events it assigns no ID to. Documentation the review found misleading: - The continuation token is opaque and produced by the state store, not by Dapr, so it must not be parsed or expected to survive a component change. - task_scheduled_id has no presence on the wire, so a 0 is equally a real event ID or a field the runtime never set. Unlike event_id there is no sentinel to test for. - The AGENTS.md "Public API" block claimed to list every exported symbol and omitted the ones this PR adds. The example no longer sleeps after start(), which already waits for the worker's stream, and its rerun line now names both spellings of the same number: the runtime's error calls it "activity task #2" where we call it "event #2". Output recaptured from a live run rather than edited by hand. Also adds a laziness test for the async iterator, whose generator semantics differ from the sync one's, and makes the test fake build the real request so it rejects what the engine would reject. Signed-off-by: Javier Aliaga --- dapr/ext/workflow/AGENTS.md | 11 +- dapr/ext/workflow/__init__.py | 2 + dapr/ext/workflow/_durabletask/aio/client.py | 6 +- dapr/ext/workflow/_durabletask/client.py | 46 ++++---- dapr/ext/workflow/aio/dapr_workflow_client.py | 21 ++-- dapr/ext/workflow/dapr_workflow_client.py | 21 ++-- dapr/ext/workflow/workflow_management.py | 42 ++++++- examples/workflow/README.md | 14 ++- examples/workflow/workflow_management.py | 13 ++- .../test_client_management_apis.py | 48 ++++---- .../ext/workflow/test_workflow_management.py | 105 ++++++++++++++---- 11 files changed, 227 insertions(+), 102 deletions(-) diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 8fbe628ba..bb7feb964 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -88,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: @@ -141,10 +146,10 @@ 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_instances(*, page_size, continuation_token)` → `WorkflowInstanceIdPage` -- `iter_workflow_instances(*, page_size=1024)` → iterator over instance IDs, paging internally (`async for` on the async client) +- `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 one `overwriteInput` flag on the wire; the sentinel default (`client.UNSET`) is what keeps it a single argument. +- `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 `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. diff --git a/dapr/ext/workflow/__init__.py b/dapr/ext/workflow/__init__.py index 626f5ec05..7bae529cf 100644 --- a/dapr/ext/workflow/__init__.py +++ b/dapr/ext/workflow/__init__.py @@ -29,6 +29,7 @@ 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, @@ -46,6 +47,7 @@ 'WorkflowHistoryEvent', 'WorkflowHistoryEventType', 'WorkflowInstanceIdPage', + 'UNSET', 'when_all', 'when_any', 'alternate_name', diff --git a/dapr/ext/workflow/_durabletask/aio/client.py b/dapr/ext/workflow/_durabletask/aio/client.py index d9ef654bb..40d99ee5d 100644 --- a/dapr/ext/workflow/_durabletask/aio/client.py +++ b/dapr/ext/workflow/_durabletask/aio/client.py @@ -32,7 +32,6 @@ get_grpc_aio_channel, ) from dapr.ext.workflow._durabletask.client import ( - UNSET, OrchestrationStatus, TaskHubGrpcClient, TInput, @@ -41,7 +40,6 @@ WorkflowState, _new_rerun_request, _TransientTimeout, - _Unset, new_orchestration_state, ) @@ -328,7 +326,8 @@ async def rerun_orchestration_from_event( event_id: int, *, new_instance_id: Optional[str] = None, - input: Union[Any, _Unset] = UNSET, + input: Optional[Any] = None, + overwrite_input: bool = False, new_child_instance_id: Optional[str] = None, ) -> str: req = _new_rerun_request( @@ -336,6 +335,7 @@ async def rerun_orchestration_from_event( 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}.") diff --git a/dapr/ext/workflow/_durabletask/client.py b/dapr/ext/workflow/_durabletask/client.py index e70000d8a..b629cc94d 100644 --- a/dapr/ext/workflow/_durabletask/client.py +++ b/dapr/ext/workflow/_durabletask/client.py @@ -33,20 +33,6 @@ class _TransientTimeout(Exception): budget. Callers convert this to a public ``TimeoutError``.""" -class _Unset: - """Sentinel for an argument that was not supplied. - - Distinct from ``None``, which is a meaningful value for the rerun input: - omitting ``input`` keeps the original, passing ``None`` clears it. - """ - - def __repr__(self) -> str: - return '' - - -UNSET = _Unset() - - TInput = TypeVar('TInput') TOutput = TypeVar('TOutput') @@ -169,17 +155,29 @@ def _new_rerun_request( event_id: int, *, new_instance_id: Optional[str], - input: Union[Any, _Unset], + input: Optional[Any], + overwrite_input: bool, new_child_instance_id: Optional[str], ) -> pb.RerunWorkflowFromEventRequest: - """Build a RerunWorkflowFromEvent request, resolving the input sentinel. - - The wire format carries the replacement input in a non-optional - ``StringValue`` plus an ``overwriteInput`` flag, so "leave the input alone" - and "replace it with null" are only distinguishable through the flag. The - sentinel collapses that pair into a single argument. + """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. """ - overwrite_input = not isinstance(input, _Unset) + 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, @@ -490,7 +488,8 @@ def rerun_orchestration_from_event( event_id: int, *, new_instance_id: Optional[str] = None, - input: Union[Any, _Unset] = UNSET, + input: Optional[Any] = None, + overwrite_input: bool = False, new_child_instance_id: Optional[str] = None, ) -> str: req = _new_rerun_request( @@ -498,6 +497,7 @@ def rerun_orchestration_from_event( 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}.") diff --git a/dapr/ext/workflow/aio/dapr_workflow_client.py b/dapr/ext/workflow/aio/dapr_workflow_client.py index c1bc09fdd..9d56ab945 100644 --- a/dapr/ext/workflow/aio/dapr_workflow_client.py +++ b/dapr/ext/workflow/aio/dapr_workflow_client.py @@ -32,6 +32,7 @@ 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, ) @@ -311,13 +312,13 @@ async def purge_workflow(self, instance_id: str, recursive: bool = True) -> None """ return await self.__obj.purge_orchestration(instance_id, recursive) - async def list_workflow_instances( + 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_instances instead unless you + 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. @@ -336,7 +337,7 @@ async def list_workflow_instances( ) return WorkflowInstanceIdPage._from_proto(res) - async def iter_workflow_instances(self, *, page_size: int = 1024) -> AsyncIterator[str]: + 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 @@ -351,7 +352,7 @@ async def iter_workflow_instances(self, *, page_size: int = 1024) -> AsyncIterat """ continuation_token = None while True: - page = await self.list_workflow_instances( + page = await self.list_workflow_instance_ids( page_size=page_size, continuation_token=continuation_token ) for instance_id in page.instance_ids: @@ -382,7 +383,7 @@ async def rerun_workflow_from_event( event_id: int, *, new_instance_id: Optional[str] = None, - input: Any = client.UNSET, + 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. @@ -405,7 +406,9 @@ async def rerun_workflow_from_event( 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. Supplying it at all is + 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 @@ -415,6 +418,8 @@ async def rerun_workflow_from_event( 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 @@ -422,10 +427,12 @@ async def rerun_workflow_from_event( 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, + 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 60174da68..b03d3dab0 100644 --- a/dapr/ext/workflow/dapr_workflow_client.py +++ b/dapr/ext/workflow/dapr_workflow_client.py @@ -31,6 +31,7 @@ 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, ) @@ -309,13 +310,13 @@ def purge_workflow(self, instance_id: str, recursive: bool = True): """ return self.__obj.purge_orchestration(instance_id, recursive) - def list_workflow_instances( + 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_instances instead unless you + 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. @@ -334,7 +335,7 @@ def list_workflow_instances( ) return WorkflowInstanceIdPage._from_proto(res) - def iter_workflow_instances(self, *, page_size: int = 1024) -> Iterator[str]: + 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 @@ -349,7 +350,7 @@ def iter_workflow_instances(self, *, page_size: int = 1024) -> Iterator[str]: """ continuation_token = None while True: - page = self.list_workflow_instances( + page = self.list_workflow_instance_ids( page_size=page_size, continuation_token=continuation_token ) yield from page.instance_ids @@ -379,7 +380,7 @@ def rerun_workflow_from_event( event_id: int, *, new_instance_id: Optional[str] = None, - input: Any = client.UNSET, + 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. @@ -402,7 +403,9 @@ def rerun_workflow_from_event( 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. Supplying it at all is + 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 @@ -412,6 +415,8 @@ def rerun_workflow_from_event( 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 @@ -419,11 +424,13 @@ def rerun_workflow_from_event( 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, + 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/workflow_management.py b/dapr/ext/workflow/workflow_management.py index 1dee70d22..d2cb8dad5 100644 --- a/dapr/ext/workflow/workflow_management.py +++ b/dapr/ext/workflow/workflow_management.py @@ -12,7 +12,7 @@ """Return types for the workflow management APIs. -These back :meth:`DaprWorkflowClient.list_workflow_instances` and +These back :meth:`DaprWorkflowClient.list_workflow_instance_ids` and :meth:`DaprWorkflowClient.get_workflow_history`, on both the sync and the async client. """ @@ -29,15 +29,42 @@ 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: Token to pass to the next - :meth:`DaprWorkflowClient.list_workflow_instances` call, or None - when this is the last page. + 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] @@ -153,8 +180,11 @@ class WorkflowHistoryEvent: 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. - TIMER_FIRED does not carry it; the wire puts that correlation in a - different field this type does not surface. + 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. """ diff --git a/examples/workflow/README.md b/examples/workflow/README.md index 046c56178..a9ed61c9a 100644 --- a/examples/workflow/README.md +++ b/examples/workflow/README.md @@ -566,8 +566,8 @@ This example demonstrates the three workflow management APIs on `DaprWorkflowClient`, using them to recover a failed order without changing the workflow code: -- `iter_workflow_instances()` walks every instance ID for this app, paging - behind the scenes. `list_workflow_instances()` returns a single page plus a +- `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 @@ -582,7 +582,7 @@ 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_instances()` +`dapr.ext.workflow.aio.DaprWorkflowClient`, where `iter_workflow_instance_ids()` is an `async for`. ```sh @@ -605,11 +605,17 @@ The output should look like this: *** 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 +*** 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 diff --git a/examples/workflow/workflow_management.py b/examples/workflow/workflow_management.py index 50901eb81..a2801043b 100644 --- a/examples/workflow/workflow_management.py +++ b/examples/workflow/workflow_management.py @@ -16,7 +16,7 @@ the instance fails. The three management APIs then recover it without touching the workflow code: -1. ``iter_workflow_instances()`` finds the instance among this app's instances. +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 @@ -28,8 +28,6 @@ in flight at the target event is re-dispatched. """ -from time import sleep - import dapr.ext.workflow as wf wfr = wf.WorkflowRuntime() @@ -84,20 +82,23 @@ def find_charge_event_id(client: wf.DaprWorkflowClient, workflow_instance_id: st def main(): client = wf.DaprWorkflowClient() wfr.start() - sleep(5) 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_instances()) + 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}', flush=True) + 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( diff --git a/tests/ext/workflow/durabletask/test_client_management_apis.py b/tests/ext/workflow/durabletask/test_client_management_apis.py index 5ae08a54f..ccbd09993 100644 --- a/tests/ext/workflow/durabletask/test_client_management_apis.py +++ b/tests/ext/workflow/durabletask/test_client_management_apis.py @@ -19,7 +19,7 @@ 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 UNSET, TaskHubGrpcClient +from dapr.ext.workflow._durabletask.client import TaskHubGrpcClient def _sync_client() -> TaskHubGrpcClient: @@ -37,12 +37,6 @@ def _async_client() -> AsyncTaskHubGrpcClient: return client -def test_the_input_sentinel_reads_as_unset(): - """UNSET is the documented default of a public argument, so its repr shows up - in help() output, IDE hovers and tracebacks.""" - assert repr(UNSET) == '' - - def test_list_instance_ids_omits_unset_pagination_fields(): client = _sync_client() client._stub.ListInstanceIDs.return_value = pb.ListInstanceIDsResponse() @@ -111,8 +105,8 @@ def test_rerun_sends_source_instance_and_event_id(): assert req.eventID == 4 -def test_rerun_without_input_leaves_the_original_input_alone(): - """Omitting input must not set overwriteInput, or the runtime nulls the input.""" +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() @@ -123,12 +117,12 @@ def test_rerun_without_input_leaves_the_original_input_alone(): assert not req.HasField('input') -def test_rerun_with_none_input_clears_the_input(): - """None is a value, not an omission: it overwrites the input with null.""" +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) + 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 @@ -139,7 +133,9 @@ 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}) + 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 @@ -151,21 +147,33 @@ def test_rerun_with_a_falsy_input_still_overwrites(): client = _sync_client() client._stub.RerunWorkflowFromEvent.return_value = pb.RerunWorkflowFromEventResponse() - client.rerun_orchestration_from_event('instance1', 4, input=0) + 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_explicit_unset_matches_omitting_input(): +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', 4, input=UNSET) + client.rerun_orchestration_from_event('instance1', 0) - req = client._stub.RerunWorkflowFromEvent.call_args[0][0] - assert req.overwriteInput is False + assert client._stub.RerunWorkflowFromEvent.call_args[0][0].eventID == 0 def test_rerun_omits_unset_instance_ids(): @@ -229,7 +237,7 @@ async def test_async_get_instance_history_unwraps_events(): @pytest.mark.asyncio -async def test_async_rerun_resolves_the_input_sentinel_like_the_sync_client(): +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') @@ -238,7 +246,7 @@ async def test_async_rerun_resolves_the_input_sentinel_like_the_sync_client(): 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) + 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' diff --git a/tests/ext/workflow/test_workflow_management.py b/tests/ext/workflow/test_workflow_management.py index 12be61127..ab2074ab4 100644 --- a/tests/ext/workflow/test_workflow_management.py +++ b/tests/ext/workflow/test_workflow_management.py @@ -19,11 +19,12 @@ from google.protobuf import timestamp_pb2, wrappers_pb2 import dapr.ext.workflow._durabletask.internal.protos as pb -from dapr.ext.workflow._durabletask.client import UNSET +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, @@ -58,15 +59,26 @@ def rerun_orchestration_from_event( event_id, *, new_instance_id=None, - input=UNSET, + 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, } ) @@ -240,12 +252,12 @@ def test_an_explicitly_empty_token_is_still_a_token(self): self.assertEqual('', WorkflowInstanceIdPage._from_proto(res).continuation_token) -class ListWorkflowInstancesTest(unittest.TestCase): +class ListWorkflowInstanceIdsTest(unittest.TestCase): def test_passes_pagination_arguments_through(self): fake = FakeTaskHubGrpcClient() client = new_client(fake) - client.list_workflow_instances(page_size=25, continuation_token='token1') + client.list_workflow_instance_ids(page_size=25, continuation_token='token1') self.assertEqual([(25, 'token1')], fake.list_calls) @@ -253,7 +265,7 @@ def test_defaults_to_no_pagination_arguments(self): fake = FakeTaskHubGrpcClient() client = new_client(fake) - client.list_workflow_instances() + client.list_workflow_instance_ids() self.assertEqual([(None, None)], fake.list_calls) @@ -262,14 +274,14 @@ def test_returns_the_converted_page(self): fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='next')] client = new_client(fake) - page = client.list_workflow_instances() + page = client.list_workflow_instance_ids() self.assertEqual( WorkflowInstanceIdPage(instance_ids=['a'], continuation_token='next'), page ) -class IterWorkflowInstancesTest(unittest.TestCase): +class IterWorkflowInstanceIdsTest(unittest.TestCase): def test_follows_the_continuation_token_across_pages(self): fake = FakeTaskHubGrpcClient() fake.pages = [ @@ -279,7 +291,7 @@ def test_follows_the_continuation_token_across_pages(self): ] client = new_client(fake) - self.assertEqual(['a', 'b', 'c', 'd'], list(client.iter_workflow_instances())) + self.assertEqual(['a', 'b', 'c', 'd'], list(client.iter_workflow_instance_ids())) self.assertEqual( [(1024, None), (1024, 'page2'), (1024, 'page3')], fake.list_calls, @@ -290,14 +302,14 @@ def test_stops_on_the_first_page_when_there_is_no_token(self): fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'])] client = new_client(fake) - self.assertEqual(['a'], list(client.iter_workflow_instances())) + 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_instances())) + self.assertEqual([], list(client.iter_workflow_instance_ids())) def test_keeps_paging_through_an_empty_page_that_carries_a_token(self): fake = FakeTaskHubGrpcClient() @@ -307,7 +319,7 @@ def test_keeps_paging_through_an_empty_page_that_carries_a_token(self): ] client = new_client(fake) - self.assertEqual(['a'], list(client.iter_workflow_instances())) + 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 @@ -316,7 +328,7 @@ def test_stops_on_an_empty_token_instead_of_looping_forever(self): fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='')] client = new_client(fake) - self.assertEqual(['a'], list(client.iter_workflow_instances())) + self.assertEqual(['a'], list(client.iter_workflow_instance_ids())) self.assertEqual(1, len(fake.list_calls)) def test_fetches_lazily(self): @@ -327,7 +339,7 @@ def test_fetches_lazily(self): ] client = new_client(fake) - instances = client.iter_workflow_instances() + instances = client.iter_workflow_instance_ids() next(instances) self.assertEqual(1, len(fake.list_calls)) @@ -336,7 +348,7 @@ def test_honours_the_page_size(self): fake = FakeTaskHubGrpcClient() client = new_client(fake) - list(client.iter_workflow_instances(page_size=10)) + list(client.iter_workflow_instance_ids(page_size=10)) self.assertEqual([(10, None)], fake.list_calls) @@ -373,22 +385,51 @@ def test_returns_the_new_instance_id(self): self.assertEqual('rerun1', client.rerun_workflow_from_event('instance1', 4)) - def test_omitting_input_reaches_the_engine_as_the_sentinel(self): - """None must not leak in as a default, or the runtime clears the input.""" + 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.assertIs(UNSET, fake.rerun_calls[0]['input']) + 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_an_explicit_none_input_reaches_the_engine_as_none(self): + 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() @@ -408,6 +449,7 @@ def test_forwards_every_argument(self): 'event_id': 4, 'new_instance_id': 'new1', 'input': {'amount': 10}, + 'overwrite_input': True, 'new_child_instance_id': 'child1', }, fake.rerun_calls[0], @@ -420,7 +462,7 @@ async def test_list_returns_the_converted_page(self): fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='next')] client = new_async_client(fake) - page = await client.list_workflow_instances(page_size=25) + page = await client.list_workflow_instance_ids(page_size=25) self.assertEqual( WorkflowInstanceIdPage(instance_ids=['a'], continuation_token='next'), page @@ -436,7 +478,7 @@ async def test_iter_follows_the_continuation_token_across_pages(self): client = new_async_client(fake) self.assertEqual( - ['a', 'b'], [instance_id async for instance_id in client.iter_workflow_instances()] + ['a', 'b'], [instance_id async for instance_id in client.iter_workflow_instance_ids()] ) self.assertEqual([(1024, None), (1024, 'page2')], fake.list_calls) @@ -445,11 +487,27 @@ async def test_iter_stops_on_an_empty_token_instead_of_looping_forever(self): fake.pages = [pb.ListInstanceIDsResponse(instanceIds=['a'], continuationToken='')] client = new_async_client(fake) - collected = [instance_id async for instance_id in client.iter_workflow_instances()] + 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'))] @@ -459,14 +517,14 @@ async def test_get_history_converts_every_event(self): self.assertEqual([(2, 'c')], [(e.event_id, e.name) for e in history]) - async def test_rerun_omitting_input_reaches_the_engine_as_the_sentinel(self): + 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.assertIs(UNSET, fake.rerun_calls[0]['input']) + self.assertFalse(fake.rerun_calls[0]['overwrite_input']) async def test_rerun_forwards_every_argument(self): fake = AsyncFakeTaskHubGrpcClient() @@ -486,6 +544,7 @@ async def test_rerun_forwards_every_argument(self): 'event_id': 4, 'new_instance_id': 'new1', 'input': None, + 'overwrite_input': True, 'new_child_instance_id': 'child1', }, fake.rerun_calls[0],