Add stateful-history delta delivery to the workflow worker

[JoshVanL]

durabletask-go adds the stateful-history optimization: a worker caches an instance's committed history between turns on its work-item stream, and the sidecar sends only the new events instead of the full history each turn. This implements the worker side in the vendored durabletask runtime; the backend is daprd, so there is no other production change.

Regenerate the durabletask proto stubs
(dapr/ext/workflow/_durabletask/internal) for the new contract: the CachedHistory message, optional WorkflowRequest.cachedHistory, GetWorkItemsRequest.capabilities, and WORKER_CAPABILITY_STATEFUL_HISTORY (the never-implemented HISTORY_STREAMING value is now reserved).

worker.py (TaskHubGrpcWorker):

• Add _WorkflowHistoryCache: a lock-guarded, per-stream cache of each instance's committed history, reclaimed by a sliding TTL, an instance-count cap, and an optional byte budget (LRU eviction). Eviction is always safe because a miss is recovered via the GetInstanceHistory RPC.
• Advertise WORKER_CAPABILITY_STATEFUL_HISTORY on GetWorkItems.
• Before replay, reconstruct the full history: for a delta work item (cachedHistory) prepend the cached prefix to the delta, falling back to GetInstanceHistory on any miss (cold stream, eviction, prefix-length mismatch); otherwise use the full pastEvents.
• After a turn, cache the committed history (never the new events), and drop it when the turn ends the execution (a completeWorkflow action, whatever its status); reset the whole cache on stream reconnect; sweep TTL on a janitor thread.
• Add a disable_stateful_history opt-out and history_cache_ttl / _max_instances / _max_bytes tuning to TaskHubGrpcWorker.

Requires:
dapr/durabletask-protobuf#54
dapr/durabletask-go#110
dapr/dapr#10142

[codecov]

Codecov Report

❌ Patch coverage is 64.92147% with 67 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.22%. Comparing base (f33c29f) to head (72744ed).

Files with missing lines	Patch %	Lines
.../_durabletask/internal/orchestrator_service_pb2.py	1.63%	60 Missing ⚠️
dapr/ext/workflow/_durabletask/worker.py	94.16%	7 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1112      +/-   ##
==========================================
+ Coverage   82.86%   83.22%   +0.36%     
==========================================
  Files         123      123              
  Lines       10130    10261     +131     
==========================================
+ Hits         8394     8540     +146     
+ Misses       1736     1721      -15     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

Implements the worker-side support for Durable Task’s stateful-history (delta) delivery optimization in the vendored workflow runtime. This adds a per-stream committed-history cache so daprd can omit already-known history prefixes on subsequent turns, reducing per-turn payload sizes while preserving correctness via fallback history fetches.

Changes:

• Add a lock-guarded, bounded (TTL / max instances / optional byte budget) per-stream workflow history cache and integrate it into TaskHubGrpcWorker history replay.
• Advertise WORKER_CAPABILITY_STATEFUL_HISTORY via GetWorkItemsRequest.capabilities and reconstruct full history for delta work items (fallback to GetInstanceHistory on cache miss).
• Regenerate durabletask protobuf stubs and add unit/e2e tests covering cache behavior and end-to-end correctness.

Reviewed changes

Copilot reviewed 4 out of 5 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
dapr/ext/workflow/_durabletask/worker.py	Adds history cache + janitor, capability advertisement, delta history reconstruction, and cache maintenance on turn completion.
tests/ext/workflow/durabletask/test_worker_history_cache.py	New unit tests for cache bounds/eviction and _resolve_history fallback behavior.
tests/ext/workflow/durabletask/test_orchestration_e2e.py	Adds an e2e workflow that exercises multi-turn behavior with optimization enabled/disabled.
dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi	Updated type stubs for new CachedHistory, capabilities, and capability enum changes.
dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py	Regenerated protobuf runtime module reflecting the updated contract.

Files not reviewed (1)

• dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py: Generated file

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 14 out of 15 changed files in this pull request and generated no new comments.

Files not reviewed (1)

• dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py: Generated file

Suppressed comments (2)

dapr/ext/workflow/_durabletask/worker.py:291

• _WorkflowHistoryCache.put() always computes ByteSize() for every event to maintain _total_bytes, even when max_bytes is left at the default unbounded value (0). For large histories this adds O(n) protobuf sizing work per turn even though byte-based eviction is disabled, which can significantly reduce the benefit of stateful-history caching.

Consider skipping byte-size accounting entirely when _max_bytes == 0 (e.g., store num_bytes=0 and keep _total_bytes=0), and only computing sizes when a byte budget is actually configured. This will require updating the unit test that currently asserts byte accounting on the default cache.

    def put(self, instance_id: str, events: list[pb.HistoryEvent]) -> None:
        """Caches an instance's committed history, evicting LRU entries to stay in bounds."""
        num_bytes = sum(event.ByteSize() for event in events)
        with self._lock:

.github/workflows/run-tests.yaml:134

• validate-dapr-head builds daprd from dapr/dapr master by default (commit: ... || 'master'). Because this workflow also runs on tag pushes, a release/tag build for this repo can be blocked by unrelated breakage or changes in dapr/dapr master, making the check non-deterministic.

If the intent is to gate PRs/main on this optimization but keep release/tag validation stable, consider skipping this job on tag pushes (or alternatively pinning dapr/dapr to a known-good commit by default and only using master when explicitly requested).

  validate-dapr-head:
    runs-on: ubuntu-latest
    env: