Preserve trailing external events after continue-as-new (#281)

Gate ordinary external-event delivery on the existing terminal context boundary while retaining entity routing and raw carryover buffering.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Bernd Verst <github@bernd.dev>

Author: 3 people

CHANGELOG.md

@@ -18,6 +18,10 @@ state is preserved.
 - `continue_as_new(..., save_events=True)` now preserves the global arrival
 order of unconsumed buffered external events across different event names,
 instead of grouping carryover events by name.
+- Fixed external events arriving after `continue_as_new(..., save_events=True)`
+being lost to abandoned waits instead of carried into the next execution.
+Events already delivered to live waits are not carried over, and
+`save_events=False` still discards unprocessed events.
 
 ## v1.11.0
 

azure-functions-durable/CHANGELOG.md

@@ -20,6 +20,9 @@ continue-as-new. This applies to both native and compatibility orchestration API
 - With the corresponding core `durabletask` SDK fix,
 `continue_as_new(..., save_events=True)` preserves the global arrival order
 of unconsumed buffered external events across different event names.
+- With a corrected core `durabletask` SDK, native durabletask orchestrators
+preserve external events arriving after `continue_as_new(..., save_events=True)`
+instead of losing them to abandoned waits.
 
 ## v2.0.0rc2
 

durabletask-azuremanaged/CHANGELOG.md

@@ -20,6 +20,9 @@ change that behavior.
 - With the corresponding core `durabletask` SDK fix,
 `continue_as_new(..., save_events=True)` preserves the global arrival order
 of unconsumed buffered external events across different event names.
+- With a corrected core `durabletask` SDK, external events arriving after
+`continue_as_new(..., save_events=True)` are no longer lost to abandoned waits
+and are carried into the next execution.
 
 ## v1.11.0
 

durabletask/task.py

@@ -393,12 +393,18 @@ def continue_as_new(self, new_input: Any, *, save_events: bool = False,
                         new_version: str | None = None) -> None:
         """Continue the orchestration execution as a new instance.
 
+        Orchestrators should return immediately after calling this method.
+        Subsequent external events are no longer delivered to pending waits
+        in the current execution.
+
         Parameters
         ----------
         new_input : Any
             The new input to use for the new orchestration instance.
         save_events : bool
             A flag indicating whether to add any unprocessed external events in the new orchestration history.
+            Events already delivered to a waiting task are not saved, even if
+            the orchestrator has not yielded that task.
         new_version : str | None
             An optional version to assign to the new orchestration instance.
         """

durabletask/worker.py

@@ -2852,7 +2852,9 @@ def _cancel_timer() -> None:
                         self._logger.info(f"{ctx.instance_id} Event raised: {event_name}")
                     task_list = ctx._pending_events.get(event_name, None)  # pyright: ignore[reportPrivateUsage]
                     decoded_result: Any | None = None
-                    if task_list:
+                    # Completed executions leave abandoned waits behind. Buffer
+                    # trailing events instead, so continue-as-new can carry them over.
+                    if task_list and not ctx._is_complete:  # pyright: ignore[reportPrivateUsage]
                         event_task = task_list.pop(0)
                         if not ph.is_empty(event.eventRaised.input):
                             decoded_result = self._data_converter.deserialize(
@@ -2879,7 +2881,7 @@ def _cancel_timer() -> None:
                         ctx._received_event_sequence += 1  # pyright: ignore[reportPrivateUsage]
                         if not ctx.is_replaying:
                             self._logger.info(
-                                f"{ctx.instance_id}: Event '{event_name}' has been buffered as there are no tasks waiting for it."
+                                f"{ctx.instance_id}: Event '{event_name}' has been buffered as there are no active tasks waiting for it."
                             )
             elif event.HasField("executionSuspended"):
                 if not self._is_suspended and not ctx.is_replaying:

tests/azure-functions-durable/test_worker_compat.py

@@ -542,6 +542,37 @@ def orchestrator(context):
     assert "boom" in completion.failureDetails.errorMessage
 
 
+@pytest.mark.parametrize("save_events", [True, False])
+def test_continue_as_new_preserves_trailing_events_in_worker_response(save_events: bool):
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        event_task = ctx.wait_for_external_event("event")
+        timer = ctx.create_timer(timedelta(seconds=1))
+        yield task.when_any([event_task, timer])
+        ctx.continue_as_new(None, save_events=save_events)
+
+    started_at = datetime(2026, 1, 1)
+    fire_at = started_at + timedelta(seconds=1)
+    request = pb.OrchestratorRequest(instanceId=TEST_INSTANCE_ID)
+    request.pastEvents.extend([
+        helpers.new_orchestrator_started_event(started_at),
+        helpers.new_execution_started_event("continue-events", TEST_INSTANCE_ID),
+        helpers.new_timer_created_event(1, fire_at),
+    ])
+    request.newEvents.extend([
+        helpers.new_timer_fired_event(1, fire_at),
+        helpers.new_event_raised_event("event", "1"),
+        helpers.new_event_raised_event("event", "2"),
+    ])
+    encoded = base64.b64encode(request.SerializeToString()).decode("utf-8")
+    response = _decode_orchestrator_response(
+        DurableFunctionsWorker().execute_orchestration_request(orchestrator, encoded))
+    completion = _get_completion_action(response)
+
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
+    assert [event.eventRaised.input.value for event in completion.carryoverEvents] == (
+        ["1", "2"] if save_events else [])
+
+
 def test_activity_retry_then_fan_out_uses_distinct_task_ids():
     """Regression test for Azure/azure-functions-durable-python#603."""
     def orchestrator(context):