History streaming API additions

protos/backend_service.proto

@@ -53,6 +53,9 @@ service BackendService {
     // Gets orchestration runtime state (history, etc.) for a given orchestration instance.
     rpc GetOrchestrationRuntimeState (GetOrchestrationRuntimeStateRequest) returns (GetOrchestrationRuntimeStateResponse);
 
+    // Gets the history of an orchestration instance as a stream of events.
+    rpc StreamInstanceHistory(StreamInstanceHistoryRequest) returns (stream HistoryChunk);
+
     // Completes an outstanding activity work item and adds a new event to the target orchestration's inbox.
     rpc CompleteActivityWorkItem (CompleteActivityWorkItemRequest) returns (CompleteActivityWorkItemResponse);
 
@@ -152,6 +155,10 @@ message CompleteOrchestrationWorkItemRequest {
     repeated HistoryEvent newTasks = 6;
     repeated HistoryEvent newTimers = 7;
     repeated OrchestratorMessage newMessages = 8;
+
+    // The number of work item events that were processed by the orchestrator.
+    // This field is optional. If not set, the service should assume that the orchestrator processed all events.
+    google.protobuf.Int32Value numEventsProcessed = 9;
 }
 
 // Response payload for completing an orchestration work item.

protos/orchestrator_service.proto

@@ -306,13 +306,18 @@ message OrchestratorRequest {
     repeated HistoryEvent pastEvents = 3;
     repeated HistoryEvent newEvents = 4;
     OrchestratorEntityParameters entityParameters = 5;
+    bool requiresHistoryStreaming = 6;
 }
 
 message OrchestratorResponse {
     string instanceId = 1;
     repeated OrchestratorAction actions = 2;
     google.protobuf.StringValue customStatus = 3;
     string completionToken = 4;
+
+    // The number of work item events that were processed by the orchestrator.
+    // This field is optional. If not set, the service should assume that the orchestrator processed all events.
+    google.protobuf.Int32Value numEventsProcessed = 5;
 }
 
 message CreateInstanceRequest {
@@ -639,6 +644,9 @@ service TaskHubSidecarService {
     rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse);
     rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse);
 
+    // Gets the history of an orchestration instance as a stream of events.
+    rpc StreamInstanceHistory(StreamInstanceHistoryRequest) returns (stream HistoryChunk);
+
     // Deletes and Creates the necessary resources for the orchestration service and the instance store
     rpc CreateTaskHub(CreateTaskHubRequest) returns (CreateTaskHubResponse);
 
@@ -662,6 +670,18 @@ message GetWorkItemsRequest {
     int32 maxConcurrentOrchestrationWorkItems = 1;
     int32 maxConcurrentActivityWorkItems = 2;
     int32 maxConcurrentEntityWorkItems = 3;
+
+    repeated WorkerCapability capabilities = 10;
+}
+
+enum WorkerCapability {
+    WORKER_CAPABILITY_UNSPECIFIED = 0;
+
+    // Indicates that the worker is capable of streaming instance history as a more optimized
+    // alternative to receiving the full history embedded in the orchestrator work-item.
+    // When set, the service may return work items without any history events as an optimization.
+    // It is strongly recommended that all SDKs support this capability.
+    WORKER_CAPABILITY_HISTORY_STREAMING = 1;
 }
 
 message WorkItem {
@@ -681,4 +701,16 @@ message CompleteTaskResponse {
 
 message HealthPing {
 	// No payload
+}
+
+message StreamInstanceHistoryRequest {
+    string instanceId = 1;
+    google.protobuf.StringValue executionId = 2;
+
+    // When set to true, the service may return a more optimized response suitable for workers.
+    bool forWorkItemProcessing = 3;
+}
+
+message HistoryChunk {
+    repeated HistoryEvent events = 1;
 }