Keep admitted Nexus calls pending during cold replay

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -6751,6 +6751,11 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     continue
                 ctx.logger._set_replaying(False)
                 _assert_pending_step_matches(cmd)
+                if _next_unconsumed_recorded_step() is not None:
+                    # Admission already has its durable sequence. A later task
+                    # waits for that call's terminal history instead of starting
+                    # the same operation again at a new sequence.
+                    return _state(pending)
                 pending.append(cmd)
                 return _state(pending)
             if isinstance(cmd, RecordSideEffect):

tests/fixtures/replay_regressions/nexus-pending-cancellation-no-readmission.json

@@ -0,0 +1,20 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "nexus-pending-cancellation-no-readmission",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.nexus-terminal-delivery", "input": [], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "ServiceCallStarted", "payload": {
+      "sequence": 1, "service_call_id": "original-service-call", "endpoint_name": "greeter",
+      "service_name": "shared", "operation_name": "greet", "operation_mode": "sync", "wait_for": "completed",
+      "status": "started", "service_call": {"service_call_id": "original-service-call", "target_namespace": "shared", "status": "started"}
+    }},
+    {"event_type": "CooperativeCancellationRequested", "recorded_at": "2026-10-12T07:00:00Z", "payload": {
+      "workflow_command_id": "original-cancellation", "workflow_run_id": "caller-run",
+      "cleanup_deadline_at": "2026-10-12T07:00:30Z"
+    }}
+  ],
+  "expected": {"command_sequence": []}
+}

tests/test_nexus_durable_wait.py

@@ -84,8 +84,7 @@ def test_new_nexus_call_serializes_the_leased_service_operation() -> None:
 def test_started_call_stays_pending_and_cold_replay_receives_terminal_result() -> None:
     started = _service_event("ServiceCallStarted", "started")
     pending = replay(DurableNexusCaller, [started], [], workflow_id="caller", run_id="caller-run")
-    assert len(pending.commands) == 1
-    assert isinstance(pending.commands[0], NexusServiceCall)
+    assert pending.commands == []
 
     expected = {"greeting": "hello world"}
     completed = _service_event("ServiceCallCompleted", "completed", expected)
@@ -95,6 +94,26 @@ def test_started_call_stays_pending_and_cold_replay_receives_terminal_result() -
     assert resumed.commands[0].result == expected
 
 
+def test_cancellation_request_and_cold_replays_do_not_readmit_pending_call() -> None:
+    started = _service_event("ServiceCallStarted", "started")
+    requested = {
+        "event_type": "CooperativeCancellationRequested",
+        "recorded_at": "2026-10-12T07:00:00Z",
+        "payload": {
+            "workflow_command_id": "original-cancellation",
+            "workflow_run_id": "caller-run",
+            "cleanup_deadline_at": "2026-10-12T07:00:30Z",
+        },
+    }
+    for history in ([started], [started, requested]):
+        for _ in range(3):
+            # A fresh workflow instance represents both another task and a
+            # replacement worker. Neither may allocate another admission.
+            outcome = replay(DurableNexusCaller, history, [], workflow_id="caller", run_id="caller-run")
+            assert outcome.commands == []
+            assert outcome.cancellation_delivery is None
+
+
 @workflow.defn(name="tests.nexus.catch-failure")
 class CatchingNexusCaller:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]