Acknowledge pending replay without rejected empty completions

CHANGELOG.md

@@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.4.2] - 2026-10-07
+
+### Fixed
+- Acknowledge replay waiting on an already pending timer, activity or child
+  through the existing waiting-for-history protocol. Empty command lists no
+  longer produce rejected workflow task completions.
+- Reject message-stream completion metadata without commands before sending a
+  waiting acknowledgement, preventing stream progress from being discarded.
+
 ## [2.4.1] - 2026-10-07
 
 ### Fixed

docs/sdk-reference.md

@@ -12,6 +12,13 @@ in-process `WorkflowEnvironment` test harness, and invocable activity carriers.
 
 Python workers execute server-routed query tasks after the Server advertises the query-tasks capability through cluster discovery.
 
+When replay is waiting on an already scheduled operation, the worker acknowledges
+waiting for history with its original lease and attempt. It does not schedule the
+operation again or fail the workflow. Low-level adapters calling
+`complete_workflow_task(commands=[])` use the same acknowledgement. Message-stream
+completion metadata requires nonempty commands so this path cannot discard stream
+progress.
+
 ## Install
 
 ```bash

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.4.1"
+version = "2.4.2"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.4.1"
-registry-version = "2.4.1"
+product-train = "2.4.2"
+registry-version = "2.4.2"
 supported-server-versions = "2.5.0"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/client.py

@@ -5115,11 +5115,23 @@ async def complete_workflow_task(
 
         Worker-plane endpoint, called by :class:`~durable_workflow.Worker`.
         ``commands`` is the list of serialized commands the workflow yielded
-        for this task.
+        for this task. An empty list acknowledges waiting for already scheduled
+        history without reporting a workflow failure or scheduling another operation.
         """
         if (message_stream_cursors or message_stream_waits) and not _worker_protocol_supports_message_streams():
             raise ValueError("message stream completion metadata requires worker protocol 1.15 or newer")
 
+        if not commands:
+            if message_stream_cursors or message_stream_waits:
+                raise ValueError("Message stream completion metadata requires nonempty commands.")
+            return await self.fail_workflow_task(
+                task_id=task_id,
+                lease_owner=lease_owner,
+                workflow_task_attempt=workflow_task_attempt,
+                message="Workflow task waiting for scheduled history.",
+                failure_type="WorkflowTaskWaitingForHistory",
+            )
+
         body: dict[str, Any] = {
             "lease_owner": lease_owner,
             "workflow_task_attempt": workflow_task_attempt,

tests/test_client.py

@@ -2903,6 +2903,25 @@ async def test_complete_workflow_task_matches_polyglot_fixture(self, client: Cli
         assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
         assert mock.call_args.kwargs["json"] == fixture["request"]["body"]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("metadata", [
+        {"message_stream_cursors": [{"stream_name": "input", "sequence": 7}]},
+        {"message_stream_waits": [{"stream_name": "input"}]},
+    ])
+    async def test_empty_completion_does_not_discard_message_stream_metadata(
+        self, client: Client, metadata: dict
+    ) -> None:
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock,
+                         return_value=_mock_response(200, {"completed": True})) as mock,
+            pytest.raises(ValueError, match="nonempty commands"),
+        ):
+            await client.complete_workflow_task(
+                task_id="task", lease_owner="owner", workflow_task_attempt=3,
+                commands=[], **metadata,
+            )
+        mock.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_fail_workflow_task_matches_polyglot_fixture(self, client: Client) -> None:
         fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-task-fail-parity.json"

tests/test_pending_operation_replay.py

@@ -2,10 +2,12 @@
 from __future__ import annotations
 
 from typing import Any
+from unittest.mock import AsyncMock, patch
 
+import httpx
 import pytest
 
-from durable_workflow import serializer, workflow
+from durable_workflow import Client, Worker, serializer, workflow
 from durable_workflow.errors import NonDeterministicReplayError
 from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay
 
@@ -78,6 +80,43 @@ def test_signal_replay_does_not_reschedule_pending_operation(kind: str) -> None:
         )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", OPENINGS)
+async def test_worker_acknowledges_pending_signal_replay_with_original_lease(kind: str) -> None:
+    history = initial_history() + [OPENINGS[kind], {
+        "event_type": "SignalReceived", "payload": {
+            "signal_name": "increment", "arguments": serializer.envelope([7]),
+        },
+    }]
+    requests = []
+
+    async def server(method, path, **kwargs):  # type: ignore[no-untyped-def]
+        body = kwargs["json"]
+        requests.append((method, path, body, kwargs["headers"]["Authorization"]))
+        if path.endswith("/complete") and not body["commands"]:
+            return httpx.Response(422, json={"message": "The commands field is required."},
+                                  request=httpx.Request(method, "http://test" + path))
+        assert path.endswith("/fail")
+        return httpx.Response(200, json={"outcome": "waiting_for_history", "recorded": True},
+                              request=httpx.Request(method, "http://test" + path))
+
+    async with Client("http://test", worker_token="worker-token") as client:
+        worker = Worker(client, task_queue="queue", worker_id="original-owner", workflows=[PendingOperation])
+        with patch.object(client._http, "request", new=AsyncMock(side_effect=server)):
+            commands = await worker._run_workflow_task({
+                "task_id": "original-task", "workflow_task_attempt": 4, "run_id": "original-run",
+                "workflow_type": "tests.replay.pending-operation", "history_events": history,
+                "payload_codec": "avro", "arguments": serializer.envelope([kind]),
+            })
+
+    assert commands == []
+    assert requests == [("POST", "/api/worker/workflow-tasks/original-task/fail", {
+        "lease_owner": "original-owner", "workflow_task_attempt": 4,
+        "failure": {"message": "Workflow task waiting for scheduled history.",
+                    "type": "WorkflowTaskWaitingForHistory"},
+    }, "Bearer worker-token")]
+
+
 @pytest.mark.parametrize("kind", OPENINGS)
 def test_new_operation_is_still_scheduled(kind: str) -> None:
     outcome = replay(PendingOperation, initial_history(), [kind])