Repeat the final local-activity backoff entry

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Local activities no longer retry with zero delay once their backoff list runs out. Each later retry repeats the final backoff entry, as Server-scheduled activities do (durable-workflow/.github#219).
+
 ## [2.15.0] - 2026-11-11
 
 ### Added

src/durable_workflow/worker.py

@@ -2565,7 +2565,10 @@ async def heartbeat(
                     or exception_type in non_retryable_types
                 )
                 retry = not non_retryable and attempt_number < max_attempts
-                backoff_seconds = backoff[attempt_number - 1] if retry and len(backoff) >= attempt_number else 0
+                # Once the list runs out, keep the final entry, as Server-scheduled
+                # activities do (.github#219). The list already carries the
+                # policy's maximum-interval cap.
+                backoff_seconds = backoff[min(attempt_number, len(backoff)) - 1] if retry and backoff else 0
                 if (
                     retry and command.schedule_to_close_timeout is not None
                     and time.monotonic() - started_at + backoff_seconds >= command.schedule_to_close_timeout

tests/fixtures/replay_regressions/local-activity-backoff-repeats-final-entry.json

@@ -0,0 +1,27 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "python-local-activity-backoff-repeats-final-entry",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "sdk.local-activity-backoff",
+    "input": [{"retry_policy": {"max_attempts": 5, "backoff_seconds": [1, 2]}}]
+  },
+  "command_sequence": [
+    {
+      "type": "local_activity_backoff",
+      "slept_seconds": [1, 2, 2, 2],
+      "attempt_backoff_seconds": [1, 2, 2, 2, null]
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "local_activity_backoff",
+        "slept_seconds": [1, 2, 2, 2],
+        "attempt_backoff_seconds": [1, 2, 2, 2, null]
+      }
+    ]
+  }
+}

tests/test_local_activity_backoff.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from durable_workflow import Client, Worker, activity
+from durable_workflow import worker as worker_module
+from durable_workflow.workflow import RecordLocalActivity
+
+
+async def _run_failing_local_activity(
+    monkeypatch: pytest.MonkeyPatch, policy: dict[str, object],
+) -> tuple[list[float], list[dict[str, object]]]:
+    slept: list[float] = []
+
+    async def record_sleep(seconds: float) -> None:
+        slept.append(seconds)
+
+    monkeypatch.setattr(worker_module.asyncio, "sleep", record_sleep)
+
+    @activity.defn(name="tests.local.always-fails")
+    async def always_fails() -> None:
+        raise RuntimeError("still failing")
+
+    async with Client("http://test", worker_token="worker") as client:
+        worker = Worker(client, task_queue="queue", activities=[always_fails])
+        worker._renew_local_workflow_lease = AsyncMock()  # type: ignore[method-assign]
+        command = RecordLocalActivity("tests.local.always-fails", [], retry_policy=policy)
+        await worker._execute_local_activity(
+            {"task_id": "task", "workflow_task_attempt": 1}, command, payload_codec="avro",
+        )
+    assert command.outcome is not None
+    return slept, list(command.outcome["attempts"])
+
+
+async def test_local_retries_repeat_the_final_backoff_entry(monkeypatch: pytest.MonkeyPatch) -> None:
+    slept, attempts = await _run_failing_local_activity(
+        monkeypatch, {"max_attempts": 5, "backoff_seconds": [1, 2]},
+    )
+    # Before .github#219 the third and fourth retries used a zero delay.
+    assert slept == [1, 2, 2, 2]
+    assert [attempt.get("backoff_seconds") for attempt in attempts] == [1, 2, 2, 2, None]
+
+
+async def test_local_retries_without_a_backoff_list_do_not_wait(monkeypatch: pytest.MonkeyPatch) -> None:
+    slept, attempts = await _run_failing_local_activity(monkeypatch, {"max_attempts": 3})
+    assert slept == []
+    assert len(attempts) == 3

tests/test_replay_regression_corpus.py

@@ -10,7 +10,7 @@
 
 import pytest
 
-from durable_workflow import Replayer, Worker, serializer, workflow
+from durable_workflow import Replayer, Worker, activity, serializer, workflow
 from durable_workflow.client import PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowStreamAppendItem
 from durable_workflow.errors import (
     DurableWorkflowError,
@@ -502,6 +502,47 @@
     assert actual == expected, f"{context} expected {expected!r}, observed {actual!r}"
 
 
+
+def _execute_local_activity_backoff(inputs: Any) -> dict[str, Any]:
+    """Run the real local-activity executor with an always-failing callback and
+    report the delays it waited between attempts (.github#219)."""
+    from durable_workflow import worker as worker_module
+    from durable_workflow.workflow import RecordLocalActivity
+
+    assert isinstance(inputs, list) and len(inputs) == 1 and isinstance(inputs[0], dict)
+    policy = inputs[0]["retry_policy"]
+
+    @activity.defn(name="tests.replay.local-always-fails")
+    async def always_fails() -> None:
+        raise RuntimeError("still failing")
+
+    async def run() -> dict[str, Any]:
+        slept: list[float] = []
+        original_sleep = worker_module.asyncio.sleep
+
+        async def record_sleep(seconds: float) -> None:
+            slept.append(seconds)
+
+        worker_module.asyncio.sleep = record_sleep  # type: ignore[assignment]
+        try:
+            async with Client("http://regression-corpus", worker_token="worker") as client:
+                worker = Worker(client, task_queue="queue", activities=[always_fails])
+                worker._renew_local_workflow_lease = AsyncMock()  # type: ignore[method-assign]
+                command = RecordLocalActivity("tests.replay.local-always-fails", [], retry_policy=policy)
+                await worker._execute_local_activity(
+                    {"task_id": "task", "workflow_task_attempt": 1}, command, payload_codec="avro",
+                )
+        finally:
+            worker_module.asyncio.sleep = original_sleep  # type: ignore[assignment]
+        assert command.outcome is not None
+        return {
+            "type": "local_activity_backoff",
+            "slept_seconds": slept,
+            "attempt_backoff_seconds": [attempt.get("backoff_seconds") for attempt in command.outcome["attempts"]],
+        }
+
+    return asyncio.run(run())
+
 def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]:
     assert fixture.get("fixture_schema") == FIXTURE_SCHEMA
     assert "python" in fixture.get("bindings", [])
@@ -512,6 +553,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.local-activity-backoff":
+        commands = [_execute_local_activity_backoff(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 in {
         "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch", "sdk.generator-side-effect-query-task-dispatch",
     }:
@@ -735,7 +781,7 @@
     assert isinstance(expected, dict) and expected
     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.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", "sdk.local-activity-backoff",
     }:
         _execute_fixture(fixture)
         return