Accept sticky history suffixes only when they continue the cache (#300)

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Workers with a sticky cache (`sticky_cache_capacity > 0`) advertise `sticky_history_suffix` (#300, server#1017). On Server 3.5.0 or later, a sticky hit receives only the history after the sequence the worker cached, instead of the full history on every workflow task. A suffix is used only when it continues the cached history exactly; otherwise the worker drops the entry and refetches full history.
+
 ## [2.15.2] - 2026-11-11
 
 ### Fixed

src/durable_workflow/worker.py

@@ -430,6 +430,18 @@ def _signal_arguments_envelope_from_export(
     return envelope
 
 
+def _suffix_continues(events: list[dict[str, Any]], after: int, last: Any) -> bool:
+    """Whether ``events`` run gap-free from ``after + 1`` and end at ``last``."""
+    if type(last) is not int or last < after:
+        return False
+    if any(not isinstance(event, dict) for event in events):
+        return False
+    sequences = [event.get("sequence") for event in events]
+    if any(type(sequence) is not int for sequence in sequences):
+        return False
+    return sequences == list(range(after + 1, last + 1))
+
+
 def _history_events_from_query_export(history: Any, history_export: Any) -> Any:
     if isinstance(history, list) and history:
         return history
@@ -1215,6 +1227,7 @@ def __init__(
         self._poller_tasks: set[asyncio.Task[Any]] = set()
         self._registered = False
         self._deregistration_fenced = False
+        self._sticky_history_suffix_supported = False
         self._registration_token: str | None = None
         self._registration_worker_id: str | None = None
         self._registration_namespace: object = None
@@ -1383,6 +1396,14 @@ async def _register(self) -> None:
             not isinstance(sticky_support, Mapping) or sticky_support.get("supported") is not True
         ):
             raise RuntimeError("sticky execution requires Server's advertised sticky_execution support")
+        # Server #1017: a sticky worker may take only the history after its cached sequence.
+        suffix_support = (
+            server_capabilities.get("sticky_history_suffix") if isinstance(server_capabilities, Mapping) else None
+        )
+        self._sticky_history_suffix_supported = (
+            self._sticky_cache.enabled and isinstance(suffix_support, Mapping)
+            and suffix_support.get("supported") is True and suffix_support.get("version") == 1
+        )
         self._cooperative_cancellation_supported = (
             "cooperative_cancellation" in self.capabilities
             and isinstance(protocol, Mapping)
@@ -1485,7 +1506,9 @@ async def _register(self) -> None:
             capabilities.append("prepared_local_activity_cancellation_policies")
         if PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST["worker_sessions"]["supported"]:
             capabilities.append("worker_sessions")
-        affinity_manifest = {name: dict(entry) for name, entry in PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST.items()}
+        affinity_manifest: dict[str, dict[str, Any]] = {
+            name: dict(entry) for name, entry in PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST.items()
+        }
         if self._sticky_cache.enabled:
             if "sticky_execution" not in capabilities:
                 capabilities.append("sticky_execution")
@@ -1494,6 +1517,11 @@ async def _register(self) -> None:
                 "minimum_protocol_version": "1.18",
                 "implementation": "bounded_durable_history_cache",
             }
+            if "sticky_history_suffix" not in capabilities:
+                capabilities.append("sticky_history_suffix")
+            affinity_manifest["sticky_history_suffix"] = {
+                "supported": True, "minimum_protocol_version": "1.18", "version": 1,
+            }
 
         ack = await self.client.register_worker(
             worker_id=self.worker_id,
@@ -1683,6 +1711,9 @@ async def _load_workflow_claim_history(
     ) -> list[dict[str, Any]]:
         if first_page_token is not None:
             return await self._fetch_workflow_claim_history_pages(task, [], first_page_token)
+        after = task.get("history_after_sequence")
+        if type(after) is int and after > 0:
+            return await self._load_suffix_history(task, after)
         inline = list(task.get("history_events", []))
         key = self._sticky_cache_key(task)
         mode = task.get("sticky_replay_mode")
@@ -1735,6 +1766,41 @@ async def _load_workflow_claim_history(
             return history
         return await self._fetch_workflow_claim_history_pages(task, inline, task.get("next_history_page_token"))
 
+    async def _load_suffix_history(self, task: dict[str, Any], after: int) -> list[dict[str, Any]]:
+        """Replay a sticky_history_suffix v1 answer (server#1017) or refetch full history.
+
+        A suffix is used only on a sticky hit whose cached history ends at exactly
+        ``after`` and whose events run gap-free from ``after + 1`` to
+        ``history_last_sequence``. It is never treated as a complete history.
+        """
+        key = self._sticky_cache_key(task)
+        suffix = list(task.get("history_events", []))
+        token = task.get("next_history_page_token")
+        if token is not None:
+            suffix = await self._fetch_workflow_claim_history_pages(task, suffix, token)
+        last = task.get("history_last_sequence")
+        cached = (
+            self._sticky_cache.lookup(key)
+            if key is not None and self._sticky_history_suffix_supported
+            and task.get("sticky_replay_mode") == "sticky_hit_expected" else None
+        )
+        if cached is not None:
+            prefix = cached[0]
+            if prefix and prefix[-1].get("sequence") == after and _suffix_continues(suffix, after, last):
+                history = prefix + suffix
+                if complete_history(history):
+                    task["_sticky_resume_token"] = None
+                    task["_sticky_resume_offset"] = 0
+                    self._sticky_cache.record_replay(hit=True)
+                    return history
+        if key is not None:
+            self._sticky_cache.discard(key)
+            self._sticky_cache.record_replay(hit=False, forced=True)
+        history = await self._fetch_workflow_claim_history_pages(task, [], "MA==")
+        if not complete_history(history):
+            raise LocalActivityExecutionAborted("cold replay lacks a complete canonical history prefix")
+        return history
+
     async def _fetch_workflow_claim_history_pages(
         self, task: dict[str, Any], history: list[dict[str, Any]], token: Any,
     ) -> list[dict[str, Any]]:
@@ -1782,12 +1848,16 @@ def _sticky_cache_claim(
             resume_offset=task.get("_sticky_resume_offset", 0),
         ):
             return None
-        return {
+        claim: dict[str, Any] = {
             "worker_id": self.worker_id, "workflow_id": key[0], "run_id": key[1], "build_id": key[2],
             "ttl_seconds": self._sticky_cache.ttl_seconds,
             "metrics": {name: self._sticky_cache.metrics()[name]
                         for name in ("hit", "miss", "eviction", "forced_cold_replay")},
         }
+        if self._sticky_history_suffix_supported:
+            # sticky_history_suffix v1: the Server sends the next suffix after exactly this sequence.
+            claim["history_last_sequence"] = history[-1]["sequence"]
+        return claim
 
     def sticky_cache_metrics(self) -> dict[str, int]:
         """Return replay counters, retained entries and encoded history bytes."""

tests/fixtures/replay_regressions/sticky-history-suffix-gap-refetches.json

@@ -0,0 +1,48 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "python-sticky-history-suffix-gap-refetches",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "sdk.sticky-suffix",
+    "input": [
+      {"cached_through": 4, "after": 4, "last": 6, "suffix": [5, 6]},
+      {"cached_through": 4, "after": 4, "last": 7, "suffix": [6, 7]}
+    ]
+  },
+  "command_sequence": [
+    {
+      "type": "sticky_suffix_history",
+      "history_sequences": [1, 2, 3, 4, 5, 6],
+      "authoritative_fetch_tokens": [],
+      "hit": 1,
+      "forced_cold_replay": 0
+    },
+    {
+      "type": "sticky_suffix_history",
+      "history_sequences": [1, 2, 3, 4, 5, 6, 7],
+      "authoritative_fetch_tokens": ["MA=="],
+      "hit": 0,
+      "forced_cold_replay": 1
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "sticky_suffix_history",
+        "history_sequences": [1, 2, 3, 4, 5, 6],
+        "authoritative_fetch_tokens": [],
+        "hit": 1,
+        "forced_cold_replay": 0
+      },
+      {
+        "type": "sticky_suffix_history",
+        "history_sequences": [1, 2, 3, 4, 5, 6, 7],
+        "authoritative_fetch_tokens": ["MA=="],
+        "hit": 0,
+        "forced_cold_replay": 1
+      }
+    ]
+  }
+}

tests/test_replay_regression_corpus.py

@@ -543,6 +543,46 @@
 
     return asyncio.run(run())
 
+def _execute_sticky_suffix(inputs: Any) -> list[dict[str, Any]]:
+    """Feed sticky_history_suffix v1 answers (server#1017) to the real worker over a warm
+    cache and report what it replays and which authoritative pages it fetched."""
+    assert isinstance(inputs, list) and inputs and all(isinstance(item, dict) for item in inputs)
+
+    def events(sequences: list[int]) -> list[dict[str, Any]]:
+        return [{"id": f"event-{sequence}", "sequence": sequence,
+                 "event_type": "WorkflowStarted" if sequence == 1 else "SideEffectRecorded",
+                 "payload": {"value": {"number": sequence}}} for sequence in sequences]
+
+    async def run(scenario: dict[str, Any]) -> dict[str, Any]:
+        last = scenario["last"]
+        client = AsyncMock(spec=Client)
+        client.workflow_task_history.return_value = {
+            "history_events": events(list(range(1, last + 1))), "next_history_page_token": None,
+        }
+        worker = Worker(client, task_queue="queue", worker_id="holder", build_id="build", sticky_cache_capacity=1)
+        worker._sticky_history_suffix_supported = True
+        assert worker._sticky_cache.remember(
+            ("workflow", "run", "build"), events(list(range(1, scenario["cached_through"] + 1))),
+        )
+        history = await worker._load_workflow_claim_history({
+            "task_id": "task", "workflow_id": "workflow", "run_id": "run", "workflow_task_attempt": 1,
+            "sticky_replay_mode": "sticky_hit_expected", "history_events": events(scenario["suffix"]),
+            "history_after_sequence": scenario["after"], "history_last_sequence": last,
+            "next_history_page_token": None,
+        })
+        metrics = worker.sticky_cache_metrics()
+        return {
+            "type": "sticky_suffix_history",
+            "history_sequences": [event["sequence"] for event in history],
+            "authoritative_fetch_tokens": [
+                call.kwargs["next_history_page_token"] for call in client.workflow_task_history.await_args_list
+            ],
+            "hit": metrics["hit"],
+            "forced_cold_replay": metrics["forced_cold_replay"],
+        }
+
+    return [asyncio.run(run(scenario)) for scenario in inputs]
+
 def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]:
     assert fixture.get("fixture_schema") == FIXTURE_SCHEMA
     assert "python" in fixture.get("bindings", [])
@@ -553,6 +593,11 @@
     assert isinstance(workflow_type, str) and workflow_type
     if workflow_type == "sdk.observed-worker-replay":
         return _execute_observed_worker_replay(fixture)
+    if workflow_type == "sdk.sticky-suffix":
+        commands = _execute_sticky_suffix(workflow.get("input"))
+        _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence")
+        _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected")
+        return commands
     if workflow_type == "sdk.local-activity-backoff":
         commands = [_execute_local_activity_backoff(workflow.get("input"))]
         _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence")
@@ -782,6 +827,7 @@
     if workflow.get("type") in {
         "sdk.replay-conformance-report", "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch",
         "sdk.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", "sdk.local-activity-backoff",
+        "sdk.sticky-suffix",
     }:
         _execute_fixture(fixture)
         return