Add stateful-history delta delivery to the workflow worker

.github/workflows/run-tests.yaml

@@ -113,19 +113,99 @@ jobs:
           nohup ollama serve > "$RUNNER_TEMP/ollama-serve.log" 2>&1 &
           timeout 60 bash -c 'until curl -fsS http://127.0.0.1:11434/api/tags > /dev/null; do sleep 1; done'
           ollama pull llama3.2:latest
+      # `dapr_head` covers behavior that only exists in daprd built from master;
+      # this job installs the latest release, so those tests run in
+      # validate-dapr-head instead.
       - name: Run integration tests
         run: |
-          uv run pytest tests/integration/
+          uv run pytest tests/integration/ -m "not dapr_head"
       - name: Validate examples
         run: |
           uv run pytest tests/examples/
 
+  # The stateful-history delta path (dapr/durabletask-go#110, reaching dapr via
+  # dapr/dapr#10142) is not in any release yet, so the matrix above cannot
+  # exercise it. This job builds daprd from master and runs the workflow suites
+  # against it. It also runs the vendored durabletask e2e tests, which nothing
+  # else does: build.yaml deselects `-m e2e`, and the validate job above only
+  # covers tests/integration and tests/examples.
+  validate-dapr-head:
+    runs-on: ubuntu-latest
+    env:
+      CHECKOUT_REPO: ${{ github.repository }}
+      CHECKOUT_REF: ${{ github.ref }}
+      # Distinct from the integration suite's 135xx block and the examples
+      # suite's 136xx block, and below the OS ephemeral range. See the "Port
+      # allocation" section in tests/integration/AGENTS.md.
+      E2E_HTTP_PORT: 13700
+      E2E_GRPC_PORT: 13701
+    steps:
+      - name: Parse repository_dispatch payload
+        if: github.event_name == 'repository_dispatch'
+        run: |
+          if [ ${{ github.event.client_payload.command }} = "ok-to-test" ]; then
+            echo "CHECKOUT_REPO=${{ github.event.client_payload.pull_head_repo }}" >> $GITHUB_ENV
+            echo "CHECKOUT_REF=${{ github.event.client_payload.pull_head_ref }}" >> $GITHUB_ENV
+          fi
+
+      - name: Check out code
+        uses: actions/checkout@v7
+        with:
+          repository: ${{ env.CHECKOUT_REPO }}
+          ref: ${{ env.CHECKOUT_REF }}
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.13"
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.3.0
+      - name: Install dependencies
+        run: uv sync --frozen --all-packages --group tests
+      # setup-dapr-runtime builds daprd inside its own clone of dapr/dapr, so the
+      # toolchain has to satisfy that repo's `go` directive, not this one's. The
+      # action installs Go itself only when none is on PATH, and the runner ships
+      # one that may be older than dapr master requires, so pin it here instead.
+      - name: Set up Go
+        uses: actions/setup-go@v7
+        with:
+          go-version: '1.26'
+      - name: Set up Dapr CLI
+        uses: dapr/.github/.github/actions/setup-dapr-cli@main
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+      # `dapr init` lays down the latest release, then the action overwrites
+      # ~/.dapr/bin/daprd with a build of `commit`. Placement and scheduler stay
+      # on release images, which is fine here: the whole server side of this
+      # feature lives in daprd's embedded durabletask-go grpcExecutor.
+      - name: Set up Dapr runtime from master
+        uses: dapr/.github/.github/actions/setup-dapr-runtime@main
+        with:
+          commit: ${{ github.event.inputs.daprdapr_commit || 'master' }}
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+      - name: Run stateful-history integration tests
+        run: |
+          uv run pytest tests/integration/ -m dapr_head -v
+      # These drive a bare TaskHubGrpcWorker rather than the DaprTestEnvironment
+      # fixture, so they need a sidecar of their own and take its address from
+      # DAPR_GRPC_ENDPOINT.
+      - name: Run vendored durabletask e2e tests
+        env:
+          DAPR_GRPC_ENDPOINT: 127.0.0.1:${{ env.E2E_GRPC_PORT }}
+        run: |
+          dapr run --app-id dt-e2e \
+            --dapr-grpc-port ${{ env.E2E_GRPC_PORT }} \
+            --dapr-http-port ${{ env.E2E_HTTP_PORT }} \
+            --dapr-internal-grpc-port 13702 \
+            --metrics-port 13703 &
+          timeout 90 bash -c 'until curl -fsS http://127.0.0.1:${{ env.E2E_HTTP_PORT }}/v1.0/healthz/outbound > /dev/null; do sleep 1; done'
+          uv run pytest tests/ext/workflow/durabletask -m e2e -v
+
   # Single stable check name to mark as required in branch protection, so the
   # required-checks list doesn't go stale when the Python version matrix
   # changes. Skipped required checks count as passing, hence `if: always()`
   # plus an explicit result check instead of relying on `needs` alone.
   validate-complete:
-    needs: validate
+    needs: [validate, validate-dapr-head]
     if: always()
     runs-on: ubuntu-latest
     steps:
@@ -135,3 +215,9 @@ jobs:
             echo "validate matrix result: ${{ needs.validate.result }}"
             exit 1
           fi
+      - name: Fail unless the dapr-head job succeeded
+        run: |
+          if [ "${{ needs.validate-dapr-head.result }}" != "success" ]; then
+            echo "validate-dapr-head result: ${{ needs.validate-dapr-head.result }}"
+            exit 1
+          fi

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

@@ -33,6 +33,7 @@
 )
 from dapr.ext.workflow._durabletask.client import (
     OrchestrationStatus,
+    TaskHubGrpcClient,
     TInput,
     TOutput,
     WorkflowIdReusePolicy,
@@ -229,6 +230,8 @@ async def _call_with_transient_retry(self, instance_id, timeout, call_fn):
                 code = rpc_error.code()  # type: ignore
                 if code == grpc.StatusCode.DEADLINE_EXCEEDED:
                     raise _TransientTimeout()
+                if TaskHubGrpcClient._is_deadline_cancellation(code, deadline):
+                    raise _TransientTimeout()
                 if code not in self._TRANSIENT_RPC_CODES:
                     raise
 

dapr/ext/workflow/_durabletask/client.py

@@ -307,6 +307,20 @@ def _call(grpc_timeout):
     # the long-poll), so its indefinite wait is preserved.
     _MAX_TRANSIENT_RETRY_SECONDS = 30.0
 
+    @staticmethod
+    def _is_deadline_cancellation(code, deadline: Optional[float]) -> bool:
+        """Reports whether a CANCELLED status is really the caller's deadline expiring.
+
+        As a bounded wait times out, the sidecar can reset the stream (RST_STREAM
+        CANCEL) marginally before gRPC raises DEADLINE_EXCEEDED locally, so the same
+        expiry surfaces as CANCELLED. Callers asked for a timeout and must see
+        TimeoutError either way. Requiring the budget to be spent keeps a genuine
+        cancellation, or a reset with time still on the clock, propagating as-is.
+        """
+        if code != grpc.StatusCode.CANCELLED or deadline is None:
+            return False
+        return time.monotonic() >= deadline
+
     def _call_with_transient_retry(self, instance_id, timeout, call_fn):
         """Run a gRPC wait call, retrying transient errors until the user
         timeout deadline. Re-raises non-transient errors immediately.
@@ -334,6 +348,8 @@ def _call_with_transient_retry(self, instance_id, timeout, call_fn):
                 code = rpc_error.code()  # type: ignore
                 if code == grpc.StatusCode.DEADLINE_EXCEEDED:
                     raise _TransientTimeout()
+                if self._is_deadline_cancellation(code, deadline):
+                    raise _TransientTimeout()
                 if code not in self._TRANSIENT_RPC_CODES:
                     raise
 

dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi

@@ -33,21 +33,31 @@ class _WorkerCapability:
 class _WorkerCapabilityEnumTypeWrapper(_enum_type_wrapper._EnumTypeWrapper[_WorkerCapability.ValueType], _builtins.type):
     DESCRIPTOR: _descriptor.EnumDescriptor
     WORKER_CAPABILITY_UNSPECIFIED: _WorkerCapability.ValueType  # 0
-    WORKER_CAPABILITY_HISTORY_STREAMING: _WorkerCapability.ValueType  # 1
-    """Indicates that the worker is capable of streaming instance history as a more optimized
-    alternative to receiving the full history embedded in the workflow work-item.
-    When set, the service may return work items without any history events as an optimization.
-    It is strongly recommended that all SDKs support this capability.
+    WORKER_CAPABILITY_STATEFUL_HISTORY: _WorkerCapability.ValueType  # 2
+    """Indicates that the worker retains an instance's accumulated history in
+    memory between workflow turns on the same work-item stream, so that the
+    service can send only the new events (the delta) instead of the full
+    history each turn. When the service has dispatched a turn for an
+    instance to this stream and believes the stream is still warm for it, it
+    may set WorkflowRequest.cachedHistory and drop the committed-history
+    prefix the worker already holds from pastEvents, leaving only the delta
+    there. On a cache miss the worker recovers the full history via the
+    GetInstanceHistory RPC, so the optimization never affects correctness.
     """
 
 class WorkerCapability(_WorkerCapability, metaclass=_WorkerCapabilityEnumTypeWrapper): ...
 
 WORKER_CAPABILITY_UNSPECIFIED: WorkerCapability.ValueType  # 0
-WORKER_CAPABILITY_HISTORY_STREAMING: WorkerCapability.ValueType  # 1
-"""Indicates that the worker is capable of streaming instance history as a more optimized
-alternative to receiving the full history embedded in the workflow work-item.
-When set, the service may return work items without any history events as an optimization.
-It is strongly recommended that all SDKs support this capability.
+WORKER_CAPABILITY_STATEFUL_HISTORY: WorkerCapability.ValueType  # 2
+"""Indicates that the worker retains an instance's accumulated history in
+memory between workflow turns on the same work-item stream, so that the
+service can send only the new events (the delta) instead of the full
+history each turn. When the service has dispatched a turn for an
+instance to this stream and believes the stream is still warm for it, it
+may set WorkflowRequest.cachedHistory and drop the committed-history
+prefix the worker already holds from pastEvents, leaving only the delta
+there. On a cache miss the worker recovers the full history via the
+GetInstanceHistory RPC, so the optimization never affects correctness.
 """
 Global___WorkerCapability: _TypeAlias = WorkerCapability  # noqa: Y015
 
@@ -135,6 +145,42 @@ class ActivityResponse(_message.Message):
 
 Global___ActivityResponse: _TypeAlias = ActivityResponse  # noqa: Y015
 
+@_typing.final
+class CachedHistory(_message.Message):
+    """CachedHistory is set on a WorkflowRequest when the service has intentionally
+    omitted the committed history prefix the worker is expected to already hold
+    for this instance from a previous turn on the same stream (see
+    WORKER_CAPABILITY_STATEFUL_HISTORY). Its presence means pastEvents carries
+    only the delta since the worker was last brought up to date; its absence
+    means pastEvents is the full committed history. The worker reconstructs the
+    full past history by prepending its cached events to pastEvents. The service
+    only sets this for workers that advertised
+    WORKER_CAPABILITY_STATEFUL_HISTORY and that it believes to be warm for the
+    instance, so it is always safe for a worker to fall back to the
+    GetInstanceHistory RPC.
+    """
+
+    DESCRIPTOR: _descriptor.Descriptor
+
+    EVENTCOUNT_FIELD_NUMBER: _builtins.int
+    eventCount: _builtins.int
+    """eventCount is the number of leading (committed) history events the
+    service believes the worker already holds, i.e. the length of the prefix
+    omitted from pastEvents. The worker's cached prefix must contain exactly
+    this many events; if it does not, the worker must treat this as a cache
+    miss and fetch the full history via GetInstanceHistory before applying
+    newEvents.
+    """
+    def __init__(
+        self,
+        *,
+        eventCount: _builtins.int = ...,
+    ) -> None: ...
+    _ClearFieldArgType: _TypeAlias = _typing.Literal["eventCount", b"eventCount"]  # noqa: Y015
+    def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
+
+Global___CachedHistory: _TypeAlias = CachedHistory  # noqa: Y015
+
 @_typing.final
 class WorkflowRequest(_message.Message):
     DESCRIPTOR: _descriptor.Descriptor
@@ -146,6 +192,7 @@ class WorkflowRequest(_message.Message):
     REQUIRESHISTORYSTREAMING_FIELD_NUMBER: _builtins.int
     ROUTER_FIELD_NUMBER: _builtins.int
     PROPAGATEDHISTORY_FIELD_NUMBER: _builtins.int
+    CACHEDHISTORY_FIELD_NUMBER: _builtins.int
     instanceId: _builtins.str
     requiresHistoryStreaming: _builtins.bool
     @_builtins.property
@@ -163,6 +210,14 @@ class WorkflowRequest(_message.Message):
         workflow function can access it via ctx.
         """
 
+    @_builtins.property
+    def cachedHistory(self) -> Global___CachedHistory:
+        """cachedHistory, when present, signals that pastEvents holds only the
+        delta and the worker must reconstruct the omitted prefix from its own
+        cache (or fetch it via GetInstanceHistory on a miss). Absent for
+        full-history sends.
+        """
+
     def __init__(
         self,
         *,
@@ -173,16 +228,21 @@ class WorkflowRequest(_message.Message):
         requiresHistoryStreaming: _builtins.bool = ...,
         router: _orchestration_pb2.TaskRouter | None = ...,
         propagatedHistory: _history_events_pb2.PropagatedHistory | None = ...,
+        cachedHistory: Global___CachedHistory | None = ...,
     ) -> None: ...
-    _HasFieldArgType: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "executionId", b"executionId", "propagatedHistory", b"propagatedHistory", "router", b"router"]  # noqa: Y015
+    _HasFieldArgType: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory", "_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "cachedHistory", b"cachedHistory", "executionId", b"executionId", "propagatedHistory", b"propagatedHistory", "router", b"router"]  # noqa: Y015
     def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ...
-    _ClearFieldArgType: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "executionId", b"executionId", "instanceId", b"instanceId", "newEvents", b"newEvents", "pastEvents", b"pastEvents", "propagatedHistory", b"propagatedHistory", "requiresHistoryStreaming", b"requiresHistoryStreaming", "router", b"router"]  # noqa: Y015
+    _ClearFieldArgType: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory", "_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "cachedHistory", b"cachedHistory", "executionId", b"executionId", "instanceId", b"instanceId", "newEvents", b"newEvents", "pastEvents", b"pastEvents", "propagatedHistory", b"propagatedHistory", "requiresHistoryStreaming", b"requiresHistoryStreaming", "router", b"router"]  # noqa: Y015
     def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
+    _WhichOneofReturnType__cachedHistory: _TypeAlias = _typing.Literal["cachedHistory"]  # noqa: Y015
+    _WhichOneofArgType__cachedHistory: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory"]  # noqa: Y015
     _WhichOneofReturnType__propagatedHistory: _TypeAlias = _typing.Literal["propagatedHistory"]  # noqa: Y015
     _WhichOneofArgType__propagatedHistory: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory"]  # noqa: Y015
     _WhichOneofReturnType__router: _TypeAlias = _typing.Literal["router"]  # noqa: Y015
     _WhichOneofArgType__router: _TypeAlias = _typing.Literal["_router", b"_router"]  # noqa: Y015
     @_typing.overload
+    def WhichOneof(self, oneof_group: _WhichOneofArgType__cachedHistory) -> _WhichOneofReturnType__cachedHistory | None: ...
+    @_typing.overload
     def WhichOneof(self, oneof_group: _WhichOneofArgType__propagatedHistory) -> _WhichOneofReturnType__propagatedHistory | None: ...
     @_typing.overload
     def WhichOneof(self, oneof_group: _WhichOneofArgType__router) -> _WhichOneofReturnType__router | None: ...
@@ -593,9 +653,22 @@ Global___PurgeInstancesResponse: _TypeAlias = PurgeInstancesResponse  # noqa: Y0
 class GetWorkItemsRequest(_message.Message):
     DESCRIPTOR: _descriptor.Descriptor
 
+    CAPABILITIES_FIELD_NUMBER: _builtins.int
+    @_builtins.property
+    def capabilities(self) -> _containers.RepeatedScalarFieldContainer[Global___WorkerCapability.ValueType]:
+        """capabilities advertises the optional protocol features this worker
+        supports, so the service can opt into optimizations on a per-stream
+        basis. Workers that leave this empty receive the default (fully
+        self-contained) behavior.
+        """
+
     def __init__(
         self,
+        *,
+        capabilities: _abc.Iterable[Global___WorkerCapability.ValueType] | None = ...,
     ) -> None: ...
+    _ClearFieldArgType: _TypeAlias = _typing.Literal["capabilities", b"capabilities"]  # noqa: Y015
+    def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
 
 Global___GetWorkItemsRequest: _TypeAlias = GetWorkItemsRequest  # noqa: Y015