Add workflow management APIs: list, history and rerun

dapr/ext/workflow/AGENTS.md

@@ -14,6 +14,7 @@ dapr/ext/workflow/
 ├── workflow_context.py                # WorkflowContext ABC
 ├── workflow_activity_context.py       # WorkflowActivityContext wrapper
 ├── workflow_state.py                  # WorkflowState, WorkflowStatus enum
+├── workflow_management.py             # WorkflowHistoryEvent(Type), WorkflowInstanceIdPage
 ├── retry_policy.py                    # RetryPolicy wrapper
 ├── util.py                            # gRPC address resolution
 ├── logger/options.py                  # LoggerOptions
@@ -24,6 +25,7 @@ tests/ext/workflow/
 ├── test_dapr_workflow_context.py      # Context method proxying
 ├── test_workflow_activity_context.py  # Activity context properties
 ├── test_workflow_client.py            # Sync client (mock gRPC)
+├── test_workflow_management.py        # list/history/rerun on both clients
 ├── test_workflow_client_aio.py        # Async client (IsolatedAsyncioTestCase)
 ├── test_workflow_runtime.py           # Registration, decorators, worker readiness
 ├── test_workflow_util.py              # Address resolution
@@ -86,6 +88,11 @@ from dapr.ext.workflow import (
     when_any,                 # Race combinator — wait for first task
     alternate_name,           # Decorator to set a custom registration name
     RetryPolicy,              # Retry config for activities/child workflows
+    WorkflowHistoryEvent,     # One event from an instance's execution history
+    WorkflowHistoryEventType, # Enum of history event kinds; unknown ones map to UNKNOWN
+    WorkflowInstanceIdPage,   # One page of instance IDs plus the continuation token
+    FailureDetails,           # Error carried by WorkflowHistoryEvent / TaskFailedError
+    UNSET,                    # Sentinel default for rerun_workflow_from_event's input
 )
 
 # Async client:
@@ -139,9 +146,13 @@ Client for workflow lifecycle management:
 - `terminate_workflow(instance_id, *, output, recursive)`
 - `pause_workflow(instance_id)` / `resume_workflow(instance_id)`
 - `purge_workflow(instance_id, *, recursive)`
+- `list_workflow_instance_ids(*, page_size, continuation_token)` → `WorkflowInstanceIdPage`
+- `iter_workflow_instance_ids(*, page_size=1024)` → iterator over instance IDs, paging internally (`async for` on the async client)
+- `get_workflow_history(instance_id)` → `list[WorkflowHistoryEvent]`
+- `rerun_workflow_from_event(instance_id, event_id, *, new_instance_id, input, new_child_workflow_instance_id)` → new `instance_id`. Omitting `input` keeps the original; passing `None` clears it. That pair is an `input` + `overwriteInput` pair on the wire, which the engine layer takes as-is; the public method collapses it into one argument via the exported `UNSET` sentinel.
 - `close()` — close gRPC connection
 
-Converts gRPC "no such instance exists" errors to `None` returns. The async variant in `aio/` has the same API with `async` methods.
+`get_workflow_state` converts gRPC "no such instance exists" errors to a `None` return. The other methods let the error propagate, including `get_workflow_history`, which raises NOT_FOUND for a missing or purged instance. The async variant in `aio/` has the same API with `async` methods.
 
 ### DaprWorkflowContext (`dapr_workflow_context.py`)
 

dapr/ext/workflow/__init__.py

@@ -14,7 +14,7 @@
 """
 
 # Import your main classes here
-from dapr.ext.workflow._durabletask.task import TaskFailedError
+from dapr.ext.workflow._durabletask.task import FailureDetails, TaskFailedError
 from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient
 from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext, when_all, when_any
 from dapr.ext.workflow.mcp import DaprMCPClient, MCPToolDef
@@ -28,6 +28,12 @@
 )
 from dapr.ext.workflow.retry_policy import RetryPolicy
 from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext
+from dapr.ext.workflow.workflow_management import (
+    UNSET,
+    WorkflowHistoryEvent,
+    WorkflowHistoryEventType,
+    WorkflowInstanceIdPage,
+)
 from dapr.ext.workflow.workflow_runtime import WorkflowRuntime, alternate_name
 from dapr.ext.workflow.workflow_state import WorkflowState, WorkflowStatus
 
@@ -38,11 +44,16 @@
     'WorkflowActivityContext',
     'WorkflowState',
     'WorkflowStatus',
+    'WorkflowHistoryEvent',
+    'WorkflowHistoryEventType',
+    'WorkflowInstanceIdPage',
+    'UNSET',
     'when_all',
     'when_any',
     'alternate_name',
     'RetryPolicy',
     'TaskFailedError',
+    'FailureDetails',
     'PropagationScope',
     'PropagatedHistory',
     'PropagationNotFoundError',

dapr/ext/workflow/_durabletask/aio/client.py

@@ -38,6 +38,7 @@
     TOutput,
     WorkflowIdReusePolicy,
     WorkflowState,
+    _new_rerun_request,
     _TransientTimeout,
     new_orchestration_state,
 )
@@ -307,3 +308,36 @@ async def purge_orchestration(self, instance_id: str, recursive: bool = True):
         req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
         self._logger.info(f"Purging instance '{instance_id}'.")
         await self._get_stub().PurgeInstances(req)
+
+    async def list_instance_ids(
+        self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None
+    ) -> pb.ListInstanceIDsResponse:
+        req = pb.ListInstanceIDsRequest(pageSize=page_size, continuationToken=continuation_token)
+        return await self._get_stub().ListInstanceIDs(req)
+
+    async def get_instance_history(self, instance_id: str) -> list[pb.HistoryEvent]:
+        req = pb.GetInstanceHistoryRequest(instanceId=instance_id)
+        res: pb.GetInstanceHistoryResponse = await self._get_stub().GetInstanceHistory(req)
+        return list(res.events)
+
+    async def rerun_orchestration_from_event(
+        self,
+        instance_id: str,
+        event_id: int,
+        *,
+        new_instance_id: Optional[str] = None,
+        input: Optional[Any] = None,
+        overwrite_input: bool = False,
+        new_child_instance_id: Optional[str] = None,
+    ) -> str:
+        req = _new_rerun_request(
+            instance_id,
+            event_id,
+            new_instance_id=new_instance_id,
+            input=input,
+            overwrite_input=overwrite_input,
+            new_child_instance_id=new_child_instance_id,
+        )
+        self._logger.info(f"Rerunning instance '{instance_id}' from event {event_id}.")
+        res: pb.RerunWorkflowFromEventResponse = await self._get_stub().RerunWorkflowFromEvent(req)
+        return res.newInstanceID

dapr/ext/workflow/_durabletask/client.py

@@ -150,6 +150,46 @@ def new_orchestration_state(
     )
 
 
+def _new_rerun_request(
+    instance_id: str,
+    event_id: int,
+    *,
+    new_instance_id: Optional[str],
+    input: Optional[Any],
+    overwrite_input: bool,
+    new_child_instance_id: Optional[str],
+) -> pb.RerunWorkflowFromEventRequest:
+    """Build a RerunWorkflowFromEvent request.
+
+    ``input`` and ``overwrite_input`` mirror the wire, which needs both: the
+    replacement input rides in a non-optional ``StringValue``, so "leave the
+    input alone" and "replace it with null" are only distinguishable through the
+    flag. Callers of the public client express that as one argument; see
+    :data:`dapr.ext.workflow.UNSET`.
+
+    Raises:
+        ValueError: If event_id is negative. The field is uint32 on the wire, so
+        protobuf would otherwise reject it with a message naming neither the
+        argument nor the reason.
+    """
+    if event_id < 0:
+        raise ValueError(
+            f'event_id must be non-negative, got {event_id}. The runtime reports -1 for '
+            'history events it assigns no ID to, and those cannot be rerun from.'
+        )
+
+    return pb.RerunWorkflowFromEventRequest(
+        sourceInstanceID=instance_id,
+        eventID=event_id,
+        newInstanceID=new_instance_id,
+        input=wrappers_pb2.StringValue(value=shared.to_json(input))
+        if overwrite_input and input is not None
+        else None,
+        overwriteInput=overwrite_input,
+        newChildWorkflowInstanceID=new_child_instance_id,
+    )
+
+
 class TaskHubGrpcClient:
     def __init__(
         self,
@@ -430,3 +470,36 @@ def purge_orchestration(self, instance_id: str, recursive: bool = True):
         req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
         self._logger.info(f"Purging instance '{instance_id}'.")
         self._stub.PurgeInstances(req)
+
+    def list_instance_ids(
+        self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None
+    ) -> pb.ListInstanceIDsResponse:
+        req = pb.ListInstanceIDsRequest(pageSize=page_size, continuationToken=continuation_token)
+        return self._stub.ListInstanceIDs(req)
+
+    def get_instance_history(self, instance_id: str) -> list[pb.HistoryEvent]:
+        req = pb.GetInstanceHistoryRequest(instanceId=instance_id)
+        res: pb.GetInstanceHistoryResponse = self._stub.GetInstanceHistory(req)
+        return list(res.events)
+
+    def rerun_orchestration_from_event(
+        self,
+        instance_id: str,
+        event_id: int,
+        *,
+        new_instance_id: Optional[str] = None,
+        input: Optional[Any] = None,
+        overwrite_input: bool = False,
+        new_child_instance_id: Optional[str] = None,
+    ) -> str:
+        req = _new_rerun_request(
+            instance_id,
+            event_id,
+            new_instance_id=new_instance_id,
+            input=input,
+            overwrite_input=overwrite_input,
+            new_child_instance_id=new_child_instance_id,
+        )
+        self._logger.info(f"Rerunning instance '{instance_id}' from event {event_id}.")
+        res: pb.RerunWorkflowFromEventResponse = self._stub.RerunWorkflowFromEvent(req)
+        return res.newInstanceID

dapr/ext/workflow/aio/dapr_workflow_client.py

@@ -16,7 +16,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Any, Optional, TypeVar, Union
+from typing import Any, AsyncIterator, Optional, TypeVar, Union
 from warnings import warn
 
 from grpc.aio import AioRpcError
@@ -31,6 +31,11 @@
 from dapr.ext.workflow.logger import Logger, LoggerOptions
 from dapr.ext.workflow.util import get_grpc_channel_options, getAddress
 from dapr.ext.workflow.workflow_context import Workflow
+from dapr.ext.workflow.workflow_management import (
+    UNSET,
+    WorkflowHistoryEvent,
+    WorkflowInstanceIdPage,
+)
 from dapr.ext.workflow.workflow_state import WorkflowState
 
 T = TypeVar('T')
@@ -306,3 +311,128 @@ async def purge_workflow(self, instance_id: str, recursive: bool = True) -> None
             recursive: The optional flag to also purge data from all child workflows.
         """
         return await self.__obj.purge_orchestration(instance_id, recursive)
+
+    async def list_workflow_instance_ids(
+        self, *, page_size: Optional[int] = None, continuation_token: Optional[str] = None
+    ) -> WorkflowInstanceIdPage:
+        """Fetches one page of workflow instance IDs for this app.
+
+        The listing is scoped to the app and namespace of the sidecar this
+        client is connected to. Use iter_workflow_instance_ids instead unless you
+        need to hold on to the continuation token yourself, for example to
+        resume paging in a later request.
+
+        Args:
+            page_size: The maximum number of instance IDs to return. Defaults
+            to leaving the limit unset, in which case how many come back is up
+            to the runtime and the state store behind it.
+            continuation_token: The token from a previous page, to start this
+            page where that one ended. Defaults to starting from the first page.
+
+        Returns:
+            A page of instance IDs, and the token for the next page if there is one.
+        """
+        res = await self.__obj.list_instance_ids(
+            page_size=page_size, continuation_token=continuation_token
+        )
+        return WorkflowInstanceIdPage._from_proto(res)
+
+    async def iter_workflow_instance_ids(self, *, page_size: int = 1024) -> AsyncIterator[str]:
+        """Iterates over every workflow instance ID for this app, paging as it goes.
+
+        Pages are fetched lazily, so abandoning the iterator early stops the
+        requests too. Iteration ends on the first page that comes back without
+        a usable continuation token.
+
+        Args:
+            page_size: The maximum number of instance IDs to fetch per request.
+
+        Yields:
+            Instance IDs, in the order the runtime returns them.
+        """
+        continuation_token = None
+        while True:
+            page = await self.list_workflow_instance_ids(
+                page_size=page_size, continuation_token=continuation_token
+            )
+            for instance_id in page.instance_ids:
+                yield instance_id
+            if not page.continuation_token:
+                return
+            continuation_token = page.continuation_token
+
+    async def get_workflow_history(self, instance_id: str) -> list[WorkflowHistoryEvent]:
+        """Fetches the full execution history of a workflow instance.
+
+        Args:
+            instance_id: The unique ID of the workflow instance to read.
+
+        Returns:
+            The instance's history events, oldest first.
+
+        Raises:
+            grpc.aio.AioRpcError: With code NOT_FOUND if no such instance exists,
+            or if it has been purged.
+        """
+        events = await self.__obj.get_instance_history(instance_id)
+        return [WorkflowHistoryEvent._from_proto(event) for event in events]
+
+    async def rerun_workflow_from_event(
+        self,
+        instance_id: str,
+        event_id: int,
+        *,
+        new_instance_id: Optional[str] = None,
+        input: Any = UNSET,
+        new_child_workflow_instance_id: Optional[str] = None,
+    ) -> str:
+        """Starts a new workflow instance that replays a completed one up to an event.
+
+        History up to event_id is replayed rather than re-executed, and
+        execution resumes from there. The source instance is left untouched.
+
+        The source instance must have reached a terminal state, must not be a
+        child workflow, and event_id must name an event the runtime can restart
+        from — a scheduled activity, a created timer, or a created child
+        workflow. get_workflow_history reports which events qualify via
+        WorkflowHistoryEvent.is_rerunnable.
+
+        Args:
+            instance_id: The unique ID of the workflow instance to rerun.
+            event_id: The WorkflowHistoryEvent.event_id to resume from. This is
+            the event's own ID, not its position in the history list.
+            new_instance_id: The ID to give the new instance. Defaults to a
+            random ID. Pass None rather than an empty string to get the default:
+            the runtime accepts '' and creates an instance whose ID is empty,
+            which it then cannot schedule reminders for.
+            input: Replacement input for the event being rerun. Omit it to keep
+            the original input; pass None to clear it. Forwarding code that has
+            to express "not supplied" can pass
+            :data:`dapr.ext.workflow.UNSET` explicitly. Supplying it at all is
+            rejected when event_id names a timer, which accepts no input.
+            new_child_workflow_instance_id: The ID to give the new child
+            workflow instance. Only accepted when event_id names a child
+            workflow creation event.
+
+        Returns:
+            The ID of the new workflow instance.
+
+        Raises:
+            ValueError: If event_id is negative, which includes the -1 the
+            runtime reports for history events it assigns no ID to.
+            grpc.aio.AioRpcError: With code INVALID_ARGUMENT if the source instance
+            is a child workflow, has not finished, or if event_id names an event
+            that rejects these arguments — a timer given an input, or a detached
+            workflow used as the starting point. NOT_FOUND if event_id names any
+            other event that cannot be rerun from, and ALREADY_EXISTS if
+            new_instance_id is already in use.
+        """
+        input_supplied = input is not UNSET
+        return await self.__obj.rerun_orchestration_from_event(
+            instance_id,
+            event_id,
+            new_instance_id=new_instance_id,
+            input=input if input_supplied else None,
+            overwrite_input=input_supplied,
+            new_child_instance_id=new_child_workflow_instance_id,
+        )