Implement send_event functionality with basic tests

Co-authored-by: berndverst <[email protected]>

Author: Copilot

durabletask/internal/helpers.py

@@ -124,6 +124,14 @@ def new_event_raised_event(name: str, encoded_input: Optional[str] = None) -> pb
     )
 
 
+def new_event_sent_event(instance_id: str, name: str, encoded_input: Optional[str] = None) -> pb.HistoryEvent:
+    return pb.HistoryEvent(
+        eventId=-1,
+        timestamp=timestamp_pb2.Timestamp(),
+        eventSent=pb.EventSentEvent(instanceId=instance_id, name=name, input=get_string_value(encoded_input))
+    )
+
+
 def new_suspend_event() -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
@@ -203,6 +211,14 @@ def new_create_sub_orchestration_action(
     ))
 
 
+def new_send_event_action(id: int, instance_id: str, event_name: str, encoded_data: Optional[str]) -> 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 is_empty(v: wrappers_pb2.StringValue):
     return v is None or v.value == ''
 

durabletask/task.py

@@ -163,6 +163,27 @@ def wait_for_external_event(self, name: str) -> Task:
         """
         pass
 
+    @abstractmethod
+    def send_event(self, instance_id: str, event_name: str, *,
+                   data: Optional[Any] = None) -> Task:
+        """Send an event to another orchestration instance.
+
+        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.
+        data : Optional[Any]
+            The optional JSON-serializable data to include with the event.
+
+        Returns
+        -------
+        Task
+            A Durable Task that completes when the event has been sent.
+        """
+        pass
+
     @abstractmethod
     def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
         """Continue the orchestration execution as a new instance.

durabletask/worker.py

@@ -852,6 +852,16 @@ def wait_for_external_event(self, name: str) -> task.Task:
             task_list.append(external_event_task)
         return external_event_task
 
+    def send_event(self, instance_id: str, event_name: str, *,
+                   data: Optional[Any] = None) -> task.Task:
+        id = self.next_sequence_number()
+        encoded_data = shared.to_json(data) if data is not None else None
+        action = ph.new_send_event_action(id, instance_id, event_name, encoded_data)
+        self._pending_actions[id] = action
+        send_event_task = task.CompletableTask()
+        self._pending_tasks[id] = send_event_task
+        return send_event_task
+
     def continue_as_new(self, new_input, *, save_events: bool = False) -> None:
         if self._is_complete:
             return
@@ -1188,6 +1198,17 @@ def process_event(
                         self._logger.info(
                             f"{ctx.instance_id}: Event '{event_name}' has been buffered as there are no tasks waiting for it."
                         )
+            elif event.HasField("eventSent"):
+                # This history event confirms that the event was successfully sent.
+                # Complete the corresponding send_event task.
+                event_id = event.eventId
+                send_event_task = ctx._pending_tasks.pop(event_id, None)
+                if send_event_task:
+                    # For send_event, we don't return any meaningful result, just completion
+                    send_event_task.complete(None)
+                    ctx.resume()
+                # Also remove the corresponding action from pending actions
+                ctx._pending_actions.pop(event_id, None)
             elif event.HasField("executionSuspended"):
                 if not self._is_suspended and not ctx.is_replaying:
                     self._logger.info(f"{ctx.instance_id}: Execution suspended.")
@@ -1304,8 +1325,8 @@ def _get_method_name_for_action(action: pb.OrchestratorAction) -> str:
         return task.get_name(task.OrchestrationContext.create_timer)
     elif action_type == "createSubOrchestration":
         return task.get_name(task.OrchestrationContext.call_sub_orchestrator)
-    # elif action_type == "sendEvent":
-    #    return task.get_name(task.OrchestrationContext.send_event)
+    elif action_type == "sendEvent":
+        return task.get_name(task.OrchestrationContext.send_event)
     else:
         raise NotImplementedError(f"Action type '{action_type}' not supported!")