Add orchestration send_event support

.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

CHANGELOG.md

@@ -9,6 +9,10 @@ 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. During
+replay, Python validates the target instance ID as well as the event name,
+intentionally detecting target changes that DurableTask.Core does not.
 - Added `emit_trace_spans=False` to `TaskHubGrpcWorker`,
 `TaskHubGrpcClient`, and `AsyncTaskHubGrpcClient` for hosts that own Durable
 Task lifecycle spans. In this propagate-only mode, W3C trace context remains

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,

durabletask/task.py

@@ -362,6 +362,32 @@ 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.
+        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
+        ----------
+        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(
+            "This OrchestrationContext implementation does not support "
+            "send_event()."
+        )
+
     @abstractmethod
     def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
         """Continue the orchestration execution as a new instance.

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)
@@ -1342,6 +1346,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 +1460,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 +1657,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
 

durabletask/worker.py

@@ -2103,6 +2103,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
@@ -2966,17 +2998,70 @@ 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.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()
+                ):
+                    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(
@@ -3290,6 +3375,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:
@@ -3299,8 +3399,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!")
 

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)