From 6fe2799133a122464e2968db2d9399285606f03a Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 15:04:24 -0600 Subject: [PATCH 1/5] Add orchestration send_event support Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 42bcd2ff-9f6f-4235-81b5-629d07ecaa30 --- CHANGELOG.md | 2 + durabletask/internal/helpers.py | 24 ++- durabletask/task.py | 20 ++ durabletask/testing/in_memory_backend.py | 31 ++- durabletask/worker.py | 103 +++++++++- tests/durabletask/test_large_payload.py | 22 ++ tests/durabletask/test_orchestration_e2e.py | 84 ++++++++ .../test_orchestration_executor.py | 188 +++++++++++++++++- 8 files changed, 457 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c2015651..7316a7c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED +- Added `OrchestrationContext.send_event()` for replay-safe, one-way event +delivery from an orchestration to another orchestration instance. - Added asynchronous `AsyncScheduledTaskClient` / `AsyncScheduleClient` and `AsyncExportHistoryClient` / `AsyncExportHistoryJobClient` APIs. Applications using `AsyncTaskHubGrpcClient` can now manage scheduled tasks and history diff --git a/durabletask/internal/helpers.py b/durabletask/internal/helpers.py index e96e88c9..897bf3b3 100644 --- a/durabletask/internal/helpers.py +++ b/durabletask/internal/helpers.py @@ -209,12 +209,17 @@ def is_entity_error_response(response: dict[str, Any]) -> bool: return "exceptionType" in response or isinstance(response.get("failureDetails"), dict) -def new_event_sent_event(event_id: int, instance_id: str, input: str): +def new_event_sent_event( + event_id: int, + instance_id: str, + input: str | None, + *, + name: str = "") -> pb.HistoryEvent: return pb.HistoryEvent( eventId=event_id, timestamp=timestamp_pb2.Timestamp(), eventSent=pb.EventSentEvent( - name="", + name=name, input=get_string_value(input), instanceId=instance_id ) @@ -322,6 +327,21 @@ def new_schedule_task_action(id: int, name: str, encoded_input: str | None, )) +def new_send_event_action( + id: int, + instance_id: str, + event_name: str, + encoded_data: str | None) -> pb.OrchestratorAction: + return pb.OrchestratorAction( + id=id, + sendEvent=pb.SendEventAction( + instance=pb.OrchestrationInstance(instanceId=instance_id), + name=event_name, + data=get_string_value(encoded_data), + ), + ) + + def new_call_entity_action(id: int, parent_instance_id: str, entity_id: EntityInstanceId, diff --git a/durabletask/task.py b/durabletask/task.py index 168966a8..4a11c06f 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -362,6 +362,26 @@ def wait_for_external_event(self, name: str, *, """ pass + def send_event(self, instance_id: str, event_name: str, *, + data: Any | None = None) -> None: + """Send an event to another orchestration instance. + + The target orchestration can receive the event using + :meth:`wait_for_external_event`. This is a one-way operation and does + not wait for the target orchestration to process the event. If the + target orchestration does not exist, the event is silently dropped. + + Parameters + ---------- + instance_id : str + The ID of the orchestration instance to send the event to. + event_name : str + The name of the event to send. Event names are case-insensitive. + data : Any | None + The optional serializable event payload. + """ + raise NotImplementedError + @abstractmethod def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None: """Continue the orchestration execution as a new instance. diff --git a/durabletask/testing/in_memory_backend.py b/durabletask/testing/in_memory_backend.py index 98766795..f8741238 100644 --- a/durabletask/testing/in_memory_backend.py +++ b/durabletask/testing/in_memory_backend.py @@ -1342,6 +1342,8 @@ def _raise_event_internal(self, instance_id: str, event_name: str, instance = self._instances.get(instance_id) if not instance: raise ValueError(f"Orchestration instance '{instance_id}' not found") + if self._is_terminal_status(instance.status): + return event = helpers.new_event_raised_event(event_name, event_data) instance.pending_events.append(event) @@ -1454,7 +1456,7 @@ def _process_action(self, instance: OrchestrationInstance, action: pb.Orchestrat elif action.HasField("createSubOrchestration"): self._process_create_sub_orchestration_action(instance, action) elif action.HasField("sendEvent"): - self._process_send_event_action(action.sendEvent) + self._process_send_event_action(instance, action) elif action.HasField("sendEntityMessage"): self._process_send_entity_message_action(instance, action) elif action.HasField("rewindOrchestration"): @@ -1651,16 +1653,35 @@ def watch() -> None: watcher_thread = threading.Thread(target=watch, daemon=True) watcher_thread.start() - def _process_send_event_action(self, send_event: pb.SendEventAction): + def _process_send_event_action( + self, + source_instance: OrchestrationInstance, + action: pb.OrchestratorAction): """Processes a send event action.""" - target_instance_id = send_event.instance.instanceId if send_event.instance else None + send_event = action.sendEvent + target_instance_id = ( + send_event.instance.instanceId + if send_event.HasField("instance") + else None + ) event_name = send_event.name - event_data = send_event.data.value if send_event.data else None + event_data = ( + send_event.data.value + if send_event.HasField("data") + else None + ) + + source_instance.history.append(helpers.new_event_sent_event( + action.id, + target_instance_id or "", + event_data, + name=event_name, + )) if target_instance_id: try: self._raise_event_internal(target_instance_id, event_name, event_data) - except Exception: + except ValueError: # Target instance may not exist - ignore pass diff --git a/durabletask/worker.py b/durabletask/worker.py index 46275362..10a7a67e 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -2081,6 +2081,38 @@ def _cancel_wait() -> None: external_event_task.set_cancel_handler(_cancel_wait) return external_event_task + def send_event(self, instance_id: str, event_name: str, *, + data: Any | None = None) -> None: + if not instance_id or not instance_id.strip(): + raise ValueError("instance_id cannot be empty") + if instance_id.startswith("@"): + raise ValueError( + "Instance IDs starting with '@' are reserved for entities; " + "use signal_entity() to send an entity operation." + ) + if not event_name or not event_name.strip(): + raise ValueError("event_name cannot be empty") + + id = self.next_sequence_number() + action = ph.new_send_event_action( + id, + instance_id, + event_name, + self._data_converter.serialize(data), + ) + self._pending_actions[id] = action + + if not self._is_replaying: + tracing.emit_event_raised_span( + event_name, + self.instance_id, + target_instance_id=instance_id, + parent_trace_context=( + self._orchestration_trace_context + or self._parent_trace_context + ), + ) + def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None: if self._is_complete: return @@ -2944,17 +2976,57 @@ def _cancel_timer() -> None: # matching the .NET worker. pass elif event.HasField("eventSent"): - # Check if this eventSent corresponds to an entity operation call after being translated to the old - # entity protocol by the Durable WebJobs extension. If so, treat this message similarly to - # entityOperationCalled and remove the pending action. Also store the entity id and event id for later action = ctx._pending_actions.pop(event.eventId, None) # pyright: ignore[reportPrivateUsage] - if action and action.HasField("sendEntityMessage"): - if action.sendEntityMessage.HasField("entityOperationCalled"): + if not action: + raise _get_non_determinism_error( + event.eventId, task.get_name(ctx.send_event) + ) + if action.HasField("sendEntityMessage"): + # Durable Functions translates entity messages to the + # legacy EventSent shape. Preserve the entity bookkeeping + # that reconstructs calls and lock requests on replay. + entity_message = action.sendEntityMessage + entity_target = None + if entity_message.HasField("entityOperationCalled"): + entity_target = entity_message.entityOperationCalled.targetInstanceId.value + elif entity_message.HasField("entityOperationSignaled"): + entity_target = entity_message.entityOperationSignaled.targetInstanceId.value + elif entity_message.HasField("entityLockRequested"): + lock_request = entity_message.entityLockRequested + if 0 <= lock_request.position < len(lock_request.lockSet): + entity_target = lock_request.lockSet[lock_request.position] + elif entity_message.HasField("entityUnlockSent"): + entity_target = entity_message.entityUnlockSent.targetInstanceId.value + + if entity_target != event.eventSent.instanceId: + raise _get_wrong_action_type_error( + event.eventId, + task.get_name(ctx.send_event), + action, + ) + + if entity_message.HasField("entityOperationCalled"): entity_id, event_id = self._parse_entity_event_sent_input(event) - ctx._entity_task_id_map[event_id] = (entity_id, action.sendEntityMessage.entityOperationCalled.operation, event.eventId) # pyright: ignore[reportPrivateUsage] - elif action.sendEntityMessage.HasField("entityLockRequested"): + ctx._entity_task_id_map[event_id] = (entity_id, entity_message.entityOperationCalled.operation, event.eventId) # pyright: ignore[reportPrivateUsage] + elif entity_message.HasField("entityLockRequested"): entity_id, event_id = self._parse_entity_event_sent_input(event) ctx._entity_lock_task_id_map[event_id] = (entity_id, event.eventId) # pyright: ignore[reportPrivateUsage] + elif not action.HasField("sendEvent"): + raise _get_wrong_action_type_error( + event.eventId, + task.get_name(ctx.send_event), + action, + ) + elif ( + action.sendEvent.name.casefold() + != event.eventSent.name.casefold() + ): + raise _get_wrong_action_name_error( + event.eventId, + method_name=task.get_name(ctx.send_event), + expected_task_name=event.eventSent.name, + actual_task_name=action.sendEvent.name, + ) else: eventType = event.WhichOneof("eventType") raise task.OrchestrationStateError( @@ -3277,8 +3349,21 @@ def _get_method_name_for_action(action: pb.OrchestratorAction) -> str: return task.get_name(task.OrchestrationContext.create_timer) case "createSubOrchestration": return task.get_name(task.OrchestrationContext.call_sub_orchestrator) - # case "sendEvent": - # return task.get_name(task.OrchestrationContext.send_event) + case "sendEvent": + return task.get_name(task.OrchestrationContext.send_event) + case "sendEntityMessage": + entity_message = action.sendEntityMessage + if entity_message.HasField("entityOperationCalled"): + return task.get_name(task.OrchestrationContext.call_entity) + if entity_message.HasField("entityOperationSignaled"): + return task.get_name(task.OrchestrationContext.signal_entity) + if ( + entity_message.HasField("entityLockRequested") + or entity_message.HasField("entityUnlockSent") + ): + return task.get_name(task.OrchestrationContext.lock_entities) + raise NotImplementedError( + "SendEntityMessage action type not supported!") case _: raise NotImplementedError(f"Action type '{action_type}' not supported!") diff --git a/tests/durabletask/test_large_payload.py b/tests/durabletask/test_large_payload.py index c819f9a1..8253151b 100644 --- a/tests/durabletask/test_large_payload.py +++ b/tests/durabletask/test_large_payload.py @@ -197,6 +197,28 @@ def test_orchestrator_response_actions_externalized(self): actual = res.actions[0].completeOrchestration.result.value assert actual.startswith(FakePayloadStore.TOKEN_PREFIX) + def test_send_event_action_data_externalized(self): + """Large orchestration-sent event payloads should be externalized.""" + store = FakePayloadStore(threshold_bytes=10) + large_data = "e" * 200 + + action = pb.OrchestratorAction( + id=1, + sendEvent=pb.SendEventAction( + instance=pb.OrchestrationInstance(instanceId="target"), + name="Approval", + data=sv(large_data), + ), + ) + res = pb.OrchestratorResponse( + instanceId="source", + actions=[action], + ) + externalize_payloads(res, store, instance_id="source") + + actual = res.actions[0].sendEvent.data.value + assert actual.startswith(FakePayloadStore.TOKEN_PREFIX) + def test_activity_response_externalized(self): """ActivityResponse.result should be externalized if large.""" store = FakePayloadStore(threshold_bytes=10) diff --git a/tests/durabletask/test_orchestration_e2e.py b/tests/durabletask/test_orchestration_e2e.py index 74fcf9ec..1454e9e1 100644 --- a/tests/durabletask/test_orchestration_e2e.py +++ b/tests/durabletask/test_orchestration_e2e.py @@ -653,6 +653,90 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert state.serialized_output == json.dumps("timed out") +def test_orchestration_sends_event_to_another_orchestration(): + event_payload = { + "approved": True, + "approver": "Ada", + } + + def receiver(ctx: task.OrchestrationContext, _): + first = yield ctx.wait_for_external_event("Approval") + second = yield ctx.wait_for_external_event("Approval") + return [first, second] + + def sender(ctx: task.OrchestrationContext, target_instance_id: str): + ctx.send_event( + target_instance_id, + "Approval", + data=event_payload, + ) + # Force a replay after the event is persisted. The EventSent history + # must remove the action so the target does not receive a duplicate. + yield ctx.create_timer(timedelta(milliseconds=50)) + return "sent" + + with worker.TaskHubGrpcWorker(host_address=HOST) as w: + w.add_orchestrator(receiver) + w.add_orchestrator(sender) + w.start() + + with client.TaskHubGrpcClient(host_address=HOST) as task_hub_client: + receiver_id = task_hub_client.schedule_new_orchestration(receiver) + receiver_state = task_hub_client.wait_for_orchestration_start( + receiver_id, timeout=30) + assert receiver_state is not None + + sender_id = task_hub_client.schedule_new_orchestration( + sender, input=receiver_id) + sender_state = task_hub_client.wait_for_orchestration_completion( + sender_id, timeout=30) + task_hub_client.raise_orchestration_event( + receiver_id, "Approval", data="client") + receiver_state = task_hub_client.wait_for_orchestration_completion( + receiver_id, timeout=30) + + assert sender_state is not None + assert sender_state.runtime_status == client.OrchestrationStatus.COMPLETED + assert sender_state.serialized_output == json.dumps("sent") + assert receiver_state is not None + assert receiver_state.runtime_status == client.OrchestrationStatus.COMPLETED + assert receiver_state.serialized_output == json.dumps( + [event_payload, "client"]) + + +def test_orchestration_send_event_drops_undeliverable_events(): + receiver_invocations = 0 + + def receiver(ctx: task.OrchestrationContext, _): + nonlocal receiver_invocations + receiver_invocations += 1 + return "completed" + + def sender(ctx: task.OrchestrationContext, target_instance_id: str): + ctx.send_event(target_instance_id, "Ignored") + ctx.send_event("missing-instance", "Ignored") + + with worker.TaskHubGrpcWorker(host_address=HOST) as w: + w.add_orchestrator(receiver) + w.add_orchestrator(sender) + w.start() + + with client.TaskHubGrpcClient(host_address=HOST) as task_hub_client: + receiver_id = task_hub_client.schedule_new_orchestration(receiver) + receiver_state = task_hub_client.wait_for_orchestration_completion( + receiver_id, timeout=30) + sender_id = task_hub_client.schedule_new_orchestration( + sender, input=receiver_id) + sender_state = task_hub_client.wait_for_orchestration_completion( + sender_id, timeout=30) + + assert receiver_state is not None + assert receiver_state.runtime_status == client.OrchestrationStatus.COMPLETED + assert sender_state is not None + assert sender_state.runtime_status == client.OrchestrationStatus.COMPLETED + assert receiver_invocations == 1 + + @pytest.mark.parametrize("winning_event", ["Approve", "Reject"]) def test_when_any_cancels_competing_external_event(winning_event: str): """Verify that the losing external-event task in a when_any race is diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 0b845671..96b52994 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -4,13 +4,14 @@ import json import logging from datetime import datetime, timedelta +from typing import Any import pytest import durabletask.internal.helpers as helpers import durabletask.internal.orchestrator_service_pb2 as pb from durabletask import task, worker, entities -from durabletask.serialization import JsonDataConverter +from durabletask.serialization import DataConverter, JsonDataConverter logging.basicConfig( format='%(asctime)s.%(msecs)03d %(name)s %(levelname)s: %(message)s', @@ -1423,6 +1424,191 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert complete_action.result.value == "42" +def test_send_event_action(): + """An orchestration can emit a one-way event action.""" + payload = {"approved": True, "approver": "Ada"} + + def orchestrator(ctx: task.OrchestrationContext, _): + result = ctx.send_event("target-instance", "Approval", data=payload) + assert result is None + return "sent" + + registry = worker._Registry() + orchestrator_name = registry.add_orchestrator(orchestrator) + new_events = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + orchestrator_name, TEST_INSTANCE_ID, encoded_input=None), + ] + + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, [], new_events) + + assert len(result.actions) == 2 + action = result.actions[0] + assert action.id == 1 + assert action.HasField("sendEvent") + assert action.sendEvent.instance.instanceId == "target-instance" + assert action.sendEvent.name == "Approval" + assert action.sendEvent.data.value == json.dumps(payload) + complete_action = result.actions[1].completeOrchestration + assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert complete_action.result.value == json.dumps("sent") + + +def test_send_event_uses_configured_data_converter(): + """Event payloads use the worker's configured data converter.""" + class UpperConverter(DataConverter): + def serialize(self, value: Any) -> str | None: + return None if value is None else json.dumps(str(value).upper()) + + def deserialize( + self, + data: str | None, + target_type: type | None = None) -> Any: + return None if data is None else json.loads(data) + + def coerce(self, value: Any, target_type: type | None = None) -> Any: + return value + + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.send_event("target-instance", "Message", data="hello") + + registry = worker._Registry() + orchestrator_name = registry.add_orchestrator(orchestrator) + new_events = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + orchestrator_name, TEST_INSTANCE_ID, encoded_input=None), + ] + + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, UpperConverter()) + result = executor.execute(TEST_INSTANCE_ID, [], new_events) + + assert result.actions[0].sendEvent.data.value == '"HELLO"' + + +@pytest.mark.parametrize( + ("instance_id", "event_name", "expected_message"), + [ + ("", "Approval", "instance_id"), + (" ", "Approval", "instance_id"), + ("@counter@1", "Approval", "reserved for entities"), + ("target-instance", "", "event_name"), + ("target-instance", " ", "event_name"), + ], +) +def test_send_event_validates_target_and_name( + instance_id: str, + event_name: str, + expected_message: str): + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.send_event(instance_id, event_name) + + registry = worker._Registry() + orchestrator_name = registry.add_orchestrator(orchestrator) + new_events = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + orchestrator_name, TEST_INSTANCE_ID, encoded_input=None), + ] + + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, [], new_events) + + complete_action = get_and_validate_complete_orchestration_action_list( + 1, result.actions) + assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert expected_message in complete_action.failureDetails.errorMessage + + +def test_send_event_replay_is_case_insensitive(): + """EventSent history removes the matching action during replay.""" + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.send_event("target-instance", "Approval", data=True) + yield ctx.wait_for_external_event("Done") + + registry = worker._Registry() + orchestrator_name = registry.add_orchestrator(orchestrator) + old_events = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + orchestrator_name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_event_sent_event( + 1, + "target-instance", + json.dumps(True), + name="approval", + ), + ] + + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute( + TEST_INSTANCE_ID, + old_events, + [helpers.new_orchestrator_started_event()], + ) + + assert result.actions == [] + + +@pytest.mark.parametrize("mismatch", ["name", "type", "missing", "entity"]) +def test_send_event_replay_detects_nondeterminism(mismatch: str): + if mismatch == "type": + def orchestrator(ctx: task.OrchestrationContext, _): + yield ctx.create_timer( + ctx.current_utc_datetime + timedelta(seconds=1)) + elif mismatch == "missing": + def orchestrator(ctx: task.OrchestrationContext, _): + yield ctx.wait_for_external_event("Done") + elif mismatch == "entity": + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.signal_entity( + entities.EntityInstanceId("Counter", "counter"), + "increment", + ) + yield ctx.wait_for_external_event("Done") + else: + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.send_event("target-instance", "ChangedName") + yield ctx.wait_for_external_event("Done") + + registry = worker._Registry() + orchestrator_name = registry.add_orchestrator(orchestrator) + old_events = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + orchestrator_name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_event_sent_event( + 1, "target-instance", None, name="OriginalName"), + ] + + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute( + TEST_INSTANCE_ID, + old_events, + [helpers.new_orchestrator_started_event()], + ) + + complete_action = result.actions[-1].completeOrchestration + assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete_action.failureDetails.errorType == ( + "durabletask.task.NonDeterminismError") + assert "send_event" in complete_action.failureDetails.errorMessage + if mismatch == "name": + assert "OriginalName" in complete_action.failureDetails.errorMessage + assert "ChangedName" in complete_action.failureDetails.errorMessage + elif mismatch == "type": + assert "create_timer" in complete_action.failureDetails.errorMessage + elif mismatch == "entity": + assert "signal_entity" in complete_action.failureDetails.errorMessage + + def test_raise_event_buffered(): """Tests that an orchestration can receive an event that arrives earlier than expected""" def orchestrator(ctx: task.OrchestrationContext, _): From 8991aa3e7a270eda28b8f14d8b51da4783caabe6 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 15:13:09 -0600 Subject: [PATCH 2/5] Require send_event context implementations Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 42bcd2ff-9f6f-4235-81b5-629d07ecaa30 --- durabletask/task.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/durabletask/task.py b/durabletask/task.py index 4a11c06f..fc680fe0 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -362,6 +362,7 @@ def wait_for_external_event(self, name: str, *, """ pass + @abstractmethod def send_event(self, instance_id: str, event_name: str, *, data: Any | None = None) -> None: """Send an event to another orchestration instance. @@ -380,7 +381,7 @@ def send_event(self, instance_id: str, event_name: str, *, data : Any | None The optional serializable event payload. """ - raise NotImplementedError + pass @abstractmethod def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None: From b1f431af93f68dc40d591627de6b9243b9ce11c4 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 21:53:40 -0600 Subject: [PATCH 3/5] Drop events queued during completion Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 42bcd2ff-9f6f-4235-81b5-629d07ecaa30 --- durabletask/testing/in_memory_backend.py | 4 ++++ tests/durabletask/test_orchestration_e2e.py | 5 +++++ 2 files changed, 9 insertions(+) diff --git a/durabletask/testing/in_memory_backend.py b/durabletask/testing/in_memory_backend.py index f8741238..0458266f 100644 --- a/durabletask/testing/in_memory_backend.py +++ b/durabletask/testing/in_memory_backend.py @@ -823,6 +823,10 @@ def CompleteOrchestratorTask(self, request: pb.OrchestratorResponse, context: gr instance.failure_details, ) ) + # Events that arrive during the final in-flight dispatch + # cannot be processed after the instance becomes terminal. + instance.pending_events.clear() + self._orchestration_queue_set.discard(request.instanceId) # Remove from in-flight before notifying or re-enqueuing self._orchestration_in_flight.discard(request.instanceId) diff --git a/tests/durabletask/test_orchestration_e2e.py b/tests/durabletask/test_orchestration_e2e.py index 1454e9e1..278f35d6 100644 --- a/tests/durabletask/test_orchestration_e2e.py +++ b/tests/durabletask/test_orchestration_e2e.py @@ -706,6 +706,7 @@ def sender(ctx: task.OrchestrationContext, target_instance_id: str): def test_orchestration_send_event_drops_undeliverable_events(): receiver_invocations = 0 + sender_invocations = 0 def receiver(ctx: task.OrchestrationContext, _): nonlocal receiver_invocations @@ -713,8 +714,11 @@ def receiver(ctx: task.OrchestrationContext, _): return "completed" def sender(ctx: task.OrchestrationContext, target_instance_id: str): + nonlocal sender_invocations + sender_invocations += 1 ctx.send_event(target_instance_id, "Ignored") ctx.send_event("missing-instance", "Ignored") + ctx.send_event(ctx.instance_id, "Ignored") with worker.TaskHubGrpcWorker(host_address=HOST) as w: w.add_orchestrator(receiver) @@ -735,6 +739,7 @@ def sender(ctx: task.OrchestrationContext, target_instance_id: str): assert sender_state is not None assert sender_state.runtime_status == client.OrchestrationStatus.COMPLETED assert receiver_invocations == 1 + assert sender_invocations == 1 @pytest.mark.parametrize("winning_event", ["Approve", "Reject"]) From ed7be01fca5b990e8b0fe9f14fefda88910976dc Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 29 Jul 2026 08:52:13 -0600 Subject: [PATCH 4/5] Preserve orchestration context compatibility Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 42bcd2ff-9f6f-4235-81b5-629d07ecaa30 --- .github/copilot-instructions.md | 11 +++++++++++ durabletask/task.py | 6 ++++-- tests/durabletask/test_orchestration_executor.py | 2 ++ 3 files changed, 17 insertions(+), 2 deletions(-) diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 9491cb2d..d5449689 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -48,6 +48,17 @@ Examples: - Follow PEP 8 conventions. - Use `autopep8` for Python formatting. +## Public API Compatibility + +- Treat adding an abstract method or property to a public ABC as a breaking + change because existing third-party subclasses will fail at instantiation. +- Do not add new abstract members outside a major release. For additive APIs in + non-major releases, provide a concrete default implementation that preserves + existing subclass instantiation and raises `NotImplementedError` only when + the new API is invoked. +- Add a regression test that confirms a new concrete API is absent from the + public ABC's `__abstractmethods__`. + ## Copyright Headers Every new Python (`.py`) source file MUST begin with the following copyright diff --git a/durabletask/task.py b/durabletask/task.py index fc680fe0..59c96684 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -362,7 +362,6 @@ def wait_for_external_event(self, name: str, *, """ pass - @abstractmethod def send_event(self, instance_id: str, event_name: str, *, data: Any | None = None) -> None: """Send an event to another orchestration instance. @@ -381,7 +380,10 @@ def send_event(self, instance_id: str, event_name: str, *, data : Any | None The optional serializable event payload. """ - pass + raise NotImplementedError( + "This OrchestrationContext implementation does not support " + "send_event()." + ) @abstractmethod def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None: diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 96b52994..623d63f1 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -1426,6 +1426,8 @@ def orchestrator(ctx: task.OrchestrationContext, _): def test_send_event_action(): """An orchestration can emit a one-way event action.""" + assert "send_event" not in task.OrchestrationContext.__abstractmethods__ + payload = {"approved": True, "approver": "Ada"} def orchestrator(ctx: task.OrchestrationContext, _): From 0072642cb91951ef35d59f8fff88648c110d25ee Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 29 Jul 2026 09:28:15 -0600 Subject: [PATCH 5/5] Detect send event target changes during replay Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 56111c85-99ce-477c-a97f-c2d57a162a55 --- CHANGELOG.md | 4 ++- durabletask/task.py | 3 ++ durabletask/worker.py | 28 +++++++++++++++++++ .../test_orchestration_executor.py | 12 +++++++- 4 files changed, 45 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7316a7c6..c1b3afd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED - Added `OrchestrationContext.send_event()` for replay-safe, one-way event -delivery from an orchestration to another orchestration instance. +delivery from an orchestration to another orchestration instance. During +replay, Python validates the target instance ID as well as the event name, +intentionally detecting target changes that DurableTask.Core does not. - Added asynchronous `AsyncScheduledTaskClient` / `AsyncScheduleClient` and `AsyncExportHistoryClient` / `AsyncExportHistoryJobClient` APIs. Applications using `AsyncTaskHubGrpcClient` can now manage scheduled tasks and history diff --git a/durabletask/task.py b/durabletask/task.py index 59c96684..edc3c39e 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -370,6 +370,9 @@ def send_event(self, instance_id: str, event_name: str, *, :meth:`wait_for_external_event`. This is a one-way operation and does not wait for the target orchestration to process the event. If the target orchestration does not exist, the event is silently dropped. + During replay, the Python SDK validates both the event name and target + instance ID. This is intentionally stricter than DurableTask.Core, + which does not validate the target instance ID. Parameters ---------- diff --git a/durabletask/worker.py b/durabletask/worker.py index 10a7a67e..9b54491e 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -3017,6 +3017,19 @@ def _cancel_timer() -> None: task.get_name(ctx.send_event), action, ) + elif ( + action.sendEvent.instance.instanceId + != event.eventSent.instanceId + ): + # Python intentionally validates the target during replay, + # unlike DurableTask.Core, to avoid silently suppressing + # delivery when orchestration code changes the target. + raise _get_wrong_action_target_error( + event.eventId, + method_name=task.get_name(ctx.send_event), + expected_instance_id=event.eventSent.instanceId, + actual_instance_id=action.sendEvent.instance.instanceId, + ) elif ( action.sendEvent.name.casefold() != event.eventSent.name.casefold() @@ -3340,6 +3353,21 @@ def _get_wrong_action_name_error( ) +def _get_wrong_action_target_error( + task_id: int, + method_name: str, + expected_instance_id: str, + actual_instance_id: str, +) -> task.NonDeterminismError: + return task.NonDeterminismError( + f"Failed to restore orchestration state due to a history mismatch: A previous execution called " + f"{method_name} with target instance ID='{expected_instance_id}' and sequence number {task_id}, but the " + f"current execution is instead targeting instance ID='{actual_instance_id}' as part of rebuilding its " + f"history. This kind of mismatch can happen if an orchestration has non-deterministic logic or if the " + f"code was changed after an instance of this orchestration already started running." + ) + + def _get_method_name_for_action(action: pb.OrchestratorAction) -> str: action_type = action.WhichOneof("orchestratorActionType") match action_type: diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 623d63f1..6e097f6c 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -1558,7 +1558,10 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert result.actions == [] -@pytest.mark.parametrize("mismatch", ["name", "type", "missing", "entity"]) +@pytest.mark.parametrize( + "mismatch", + ["name", "target", "type", "missing", "entity"], +) def test_send_event_replay_detects_nondeterminism(mismatch: str): if mismatch == "type": def orchestrator(ctx: task.OrchestrationContext, _): @@ -1574,6 +1577,10 @@ def orchestrator(ctx: task.OrchestrationContext, _): "increment", ) yield ctx.wait_for_external_event("Done") + elif mismatch == "target": + def orchestrator(ctx: task.OrchestrationContext, _): + ctx.send_event("changed-target", "OriginalName") + yield ctx.wait_for_external_event("Done") else: def orchestrator(ctx: task.OrchestrationContext, _): ctx.send_event("target-instance", "ChangedName") @@ -1605,6 +1612,9 @@ def orchestrator(ctx: task.OrchestrationContext, _): if mismatch == "name": assert "OriginalName" in complete_action.failureDetails.errorMessage assert "ChangedName" in complete_action.failureDetails.errorMessage + elif mismatch == "target": + assert "target-instance" in complete_action.failureDetails.errorMessage + assert "changed-target" in complete_action.failureDetails.errorMessage elif mismatch == "type": assert "create_timer" in complete_action.failureDetails.errorMessage elif mismatch == "entity":