Added the stream records, commands, events, and task carriage.

api-linter.yaml

@@ -12,6 +12,11 @@
   disabled_rules:
     - "core::0123::resource-annotation" # We don't require resource annotations on all messages -- https://linter.aip.dev/123/resource-annotation
 
+- included_paths:
+    - "**/stream/v1/message.proto"
+  disabled_rules:
+    - "core::0140::prepositions" # Half-open offset ranges on StreamSlice and StreamRange are spelled `from_offset`/`to_offset`, as in the server's own protos -- https://linter.aip.dev/140/prepositions
+
 - included_paths:
     - "**/workflowservice/v1/request_response.proto"
     - "**/operatorservice/v1/request_response.proto"

openapi/openapiv3.yaml

@@ -12878,6 +12878,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
           type: string
           format: enum
         version:
@@ -13043,6 +13045,10 @@ components:
           $ref: '#/components/schemas/WorkflowExecutionUnpausedEventAttributes'
         workflowExecutionTimeSkippingTransitionedEventAttributes:
           $ref: '#/components/schemas/WorkflowExecutionTimeSkippingTransitionedEventAttributes'
+        workflowStreamSubscribedEventAttributes:
+          $ref: '#/components/schemas/WorkflowStreamSubscribedEventAttributes'
+        workflowStreamRecordsAppendedEventAttributes:
+          $ref: '#/components/schemas/WorkflowStreamRecordsAppendedEventAttributes'
       description: |-
         History events are the method by which Temporal SDKs advance (or recreate) workflow state.
          See the `EventType` enum for more info about what each event is for.
@@ -14853,6 +14859,13 @@ components:
                2. Try to assign the next poll to a group without any pending polls,
                3. If every group has some pending polls, assign the next poll to a group randomly
                  according to the weights.
+        streamSlices:
+          type: array
+          items:
+            $ref: '#/components/schemas/StreamSlice'
+          description: |-
+            Stream data attached to this task. Delivered out of band so the payloads
+             never enter History; only the offset ranges are recorded there.
     PollerGroupInfo:
       type: object
       properties:
@@ -15529,6 +15542,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
           type: string
           description: The event type of the history event generated by the request.
           format: enum
@@ -17680,6 +17695,99 @@ components:
         type:
           type: string
           description: The type of the driver, required.
+    StreamRange:
+      type: object
+      properties:
+        streamId:
+          type: string
+        fromOffset:
+          type: string
+          description: Inclusive.
+        toOffset:
+          type: string
+          description: Exclusive.
+      description: |-
+        The offsets a Workflow Task consumed, without the payloads. Recorded on
+         WorkflowTaskCompleted so History grows with Workflow Tasks rather than with
+         records.
+    StreamRecord:
+      type: object
+      properties:
+        body:
+          allOf:
+            - $ref: '#/components/schemas/Payload'
+          description: |-
+            The value the producer published, stored as sent. A payload codec
+             applies here as it does to any other payload.
+        metadata:
+          type: object
+          additionalProperties:
+            $ref: '#/components/schemas/Payload'
+          description: Producer-supplied provenance, stored as sent.
+        topic:
+          type: string
+          description: Producer-supplied grouping label, stored as sent.
+        kind:
+          enum:
+            - STREAM_RECORD_KIND_UNSPECIFIED
+            - STREAM_RECORD_KIND_DATA
+            - STREAM_RECORD_KIND_FINISH
+          type: string
+          description: How to read this record. Unspecified is read as DATA.
+          format: enum
+        producerId:
+          type: string
+          description: Who wrote the record. Empty when the owning Workflow did.
+        attempt:
+          type: string
+          description: |-
+            The producer's attempt. Readers treat a later attempt by the same
+             producer as superseding what the earlier one wrote.
+        sequence:
+          type: string
+          description: |-
+            The producer's position within its attempt, or -1 when unnumbered.
+             Stored as sent; the server does not assign, validate or order by it.
+      description: |-
+        One entry in a stream. The record is the wire format: stores keep it
+         serialized as is and readers in every language decode the same bytes.
+    StreamSlice:
+      type: object
+      properties:
+        streamId:
+          type: string
+        runId:
+          type: string
+          description: |-
+            Run id of the execution that owns the stream. Set on both a slice for the
+             task being started and a re-supplied one.
+        fromOffset:
+          type: string
+          description: Inclusive.
+        toOffset:
+          type: string
+          description: |-
+            Exclusive. Equal to from_offset when the subscription observed nothing,
+             which is a fact replay has to reproduce rather than an absence of one.
+        records:
+          type: array
+          items:
+            $ref: '#/components/schemas/StreamRecord'
+        workflowTaskCompletedEventId:
+          type: string
+          description: |-
+            The WorkflowTaskCompleted event whose consumed_stream_ranges recorded
+             this range. Set only when the server is re-supplying a range for a task
+             being replayed; a slice for the task now being started leaves it unset,
+             because the event closing that task does not exist yet.
+
+             Replay needs this because a Workflow Task response carries one slice set
+             while a cache miss replays every prior task, so the ranges have to be
+             matched to the events that recorded them rather than to the response.
+      description: |-
+        A contiguous range of a stream delivered to a Workflow Task, along with the
+         offsets it covers. The offsets are what History records; the records
+         themselves are never written to History.
     StructuredCalendarSpec:
       type: object
       properties:
@@ -19830,6 +19938,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
           type: string
           format: enum
       description: EventReference is a direct reference to a history event through the event ID.
@@ -19901,6 +20011,8 @@ components:
             - EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED
             - EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED
+            - EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
+            - EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
           type: string
           format: enum
       description: RequestIdReference is a indirect reference to a history event through the request ID.
@@ -21023,6 +21135,44 @@ components:
               * BETWEEN ... AND
                 STARTS_WITH
       description: Activity trigger will be triggered when an activity is about to start.
+    WorkflowStreamRecordsAppendedEventAttributes:
+      type: object
+      properties:
+        workflowTaskCompletedEventId:
+          type: string
+          description: |-
+            The WorkflowTaskCompleted event of the task whose command appended this
+             batch.
+        streamId:
+          type: string
+          description: Stream the Workflow appended to.
+        firstOffset:
+          type: string
+          description: Offset the first record of the batch landed at.
+        recordCount:
+          type: string
+          description: |-
+            How many records the batch held. With first_offset this names the range
+             without carrying any of it, which is what keeps this event a fixed size
+             no matter how large the batch or its payloads are.
+    WorkflowStreamSubscribedEventAttributes:
+      type: object
+      properties:
+        workflowTaskCompletedEventId:
+          type: string
+          description: |-
+            The WorkflowTaskCompleted event of the task whose command created this
+             subscription.
+        streamId:
+          type: string
+          description: Stream the Workflow subscribed to.
+        startOffset:
+          type: string
+          description: |-
+            The offset the subscription actually starts from. Resolved by the server
+             when the subscription is registered and recorded here, so replay reads
+             the resolved value rather than resolving it again against a stream that
+             has since moved.
     WorkflowTaskCompletedEventAttributes:
       type: object
       properties:
@@ -21094,6 +21244,16 @@ components:
           description: |-
             The Worker Deployment Version that completed this task. Must be set if `versioning_behavior`
              is set. This value updates workflow execution's `versioning_info.deployment_version`.
+        consumedStreamRanges:
+          type: array
+          items:
+            $ref: '#/components/schemas/StreamRange'
+          description: |-
+            Offset ranges this Workflow Task consumed from streams it subscribes to.
+             Recorded on every task where a subscription is active, including when it
+             observed nothing: an empty range is a fact replay must reproduce, and
+             omitting it would let replay deliver records the Workflow did not have.
+             Numbered 20 to leave 14 through 19 free for fields added on the main line.
     WorkflowTaskCompletedMetadata:
       type: object
       properties:
@@ -21201,6 +21361,9 @@ components:
             - WORKFLOW_TASK_FAILED_CAUSE_EXTERNAL_STORAGE_FAILURE
             - WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_PAUSE_REQUESTED_BEFORE_TASK_STARTED
             - WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE
+            - WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES
+            - WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES
+            - WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE
           type: string
           format: enum
         failure:

temporal/api/command/v1/message.proto

@@ -14,6 +14,7 @@ import "google/protobuf/duration.proto";
 import "temporal/api/enums/v1/workflow.proto";
 import "temporal/api/enums/v1/command_type.proto";
 import "temporal/api/common/v1/message.proto";
+import "temporal/api/stream/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
 import "temporal/api/taskqueue/v1/message.proto";
 import "temporal/api/workflow/v1/message.proto";
@@ -324,5 +325,38 @@ message Command {
 
         ScheduleNexusOperationCommandAttributes schedule_nexus_operation_command_attributes = 18;
         RequestCancelNexusOperationCommandAttributes request_cancel_nexus_operation_command_attributes = 19;
+        AppendStreamRecordsCommandAttributes append_stream_records_command_attributes = 20;
+        SubscribeStreamCommandAttributes subscribe_stream_command_attributes = 21;
     }
 }
+
+// Appends records to a stream the Workflow owns. Applied inside the Workflow
+// Task's own commit. Produces one `WorkflowStreamRecordsAppended` event
+// carrying the offset range and none of the payload; it schedules no further
+// work.
+message AppendStreamRecordsCommandAttributes {
+    // Empty means the Workflow's default output stream.
+    string stream_id = 1;
+    // Stored in order. The server sets `producer_id` to empty on each record,
+    // because the owning Workflow is the producer here.
+    repeated temporal.api.stream.v1.StreamRecord records = 2;
+}
+
+// Subscribe this Workflow to a stream, so later Workflow Tasks carry the ranges
+// it has not consumed yet.
+//
+// The stream's addressing is resolved by the server rather than supplied here.
+// A Workflow cannot look it up without doing I/O, and a value it carried would
+// be a reading rather than a fact, so it could differ on replay.
+message SubscribeStreamCommandAttributes {
+    // Stream to consume. A stream in another execution is addressed by its id;
+    // one this Workflow owns is addressed by the name it was published under.
+    // The server resolves an owned name first and falls back to a standalone
+    // id, so a Workflow that owns a stream under this name cannot reach a
+    // standalone stream with the same id.
+    string stream_id = 1;
+    // Where to start. Negative means from wherever the stream is when the
+    // subscription is registered, which the server resolves and records so
+    // replay does not resolve it again.
+    int64 start_offset = 2;
+}

temporal/api/enums/v1/command_type.proto

@@ -29,4 +29,6 @@ enum CommandType {
     COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES = 16;
     COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION = 17;
     COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION = 18;
+    COMMAND_TYPE_APPEND_STREAM_RECORDS = 19;
+    COMMAND_TYPE_SUBSCRIBE_STREAM = 20;
 }

temporal/api/enums/v1/event_type.proto

@@ -175,4 +175,12 @@ enum EventType {
     EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED = 59;
     // An event that indicates time skipping advanced time or was disabled automatically after a bound was reached.
     EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED = 60;
+    // A Workflow subscribed to a stream. Recorded once per subscription, not
+    // per record: the offsets a task consumed ride WorkflowTaskCompleted and
+    // the payloads never enter History at all.
+    EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED = 61;
+    // A Workflow appended a batch of records to a stream. Recorded per
+    // batch, and carrying only the offset range it landed at: the bodies go to
+    // the stream's own log, never into History.
+    EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED = 62;
 }

temporal/api/enums/v1/failed_cause.proto

@@ -90,6 +90,14 @@ enum WorkflowTaskFailedCause {
     WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_PAUSE_REQUESTED_BEFORE_TASK_STARTED = 39;
     // A workflow task failed because the request exceeded a size limit.
     WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE = 40;
+    // A workflow task completed with an invalid AppendStreamRecords command.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES = 41;
+    // A workflow task completed with an invalid SubscribeStream command.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES = 42;
+    // A workflow task could not be started because a stream range it consumed and recorded in
+    // History can no longer be served, for example after truncation or because it exceeds the
+    // replay bound. Check the workflow task failure message for more information.
+    WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE = 43;
 }
 
 // Activity tasks can fail for various reasons. Note that some of these reasons can only originate

temporal/api/history/v1/message.proto

@@ -17,6 +17,7 @@ import "temporal/api/enums/v1/failed_cause.proto";
 import "temporal/api/enums/v1/update.proto";
 import "temporal/api/enums/v1/workflow.proto";
 import "temporal/api/common/v1/message.proto";
+import "temporal/api/stream/v1/message.proto";
 import "temporal/api/deployment/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
 import "temporal/api/taskqueue/v1/message.proto";
@@ -379,6 +380,13 @@ message WorkflowTaskCompletedEventAttributes {
     // The Worker Deployment Version that completed this task. Must be set if `versioning_behavior`
     // is set. This value updates workflow execution's `versioning_info.deployment_version`.
     temporal.api.deployment.v1.WorkerDeploymentVersion deployment_version = 11;
+
+    // Offset ranges this Workflow Task consumed from streams it subscribes to.
+    // Recorded on every task where a subscription is active, including when it
+    // observed nothing: an empty range is a fact replay must reproduce, and
+    // omitting it would let replay deliver records the Workflow did not have.
+    // Numbered 20 to leave 14 through 19 free for fields added on the main line.
+    repeated temporal.api.stream.v1.StreamRange consumed_stream_ranges = 20;
 }
 
 message WorkflowTaskTimedOutEventAttributes {
@@ -955,6 +963,33 @@ message ActivityPropertiesModifiedExternallyEventAttributes {
     temporal.api.common.v1.RetryPolicy new_retry_policy = 2;
 }
 
+message WorkflowStreamSubscribedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command created this
+    // subscription.
+    int64 workflow_task_completed_event_id = 1;
+    // Stream the Workflow subscribed to.
+    string stream_id = 2;
+    // The offset the subscription actually starts from. Resolved by the server
+    // when the subscription is registered and recorded here, so replay reads
+    // the resolved value rather than resolving it again against a stream that
+    // has since moved.
+    int64 start_offset = 3;
+}
+
+message WorkflowStreamRecordsAppendedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command appended this
+    // batch.
+    int64 workflow_task_completed_event_id = 1;
+    // Stream the Workflow appended to.
+    string stream_id = 2;
+    // Offset the first record of the batch landed at.
+    int64 first_offset = 3;
+    // How many records the batch held. With first_offset this names the range
+    // without carrying any of it, which is what keeps this event a fixed size
+    // no matter how large the batch or its payloads are.
+    int64 record_count = 4;
+}
+
 message WorkflowExecutionUpdateAcceptedEventAttributes {
     // The instance ID of the update protocol that generated this event.
     string protocol_instance_id = 1;
@@ -1278,6 +1313,8 @@ message HistoryEvent {
         WorkflowExecutionPausedEventAttributes workflow_execution_paused_event_attributes = 63;
         WorkflowExecutionUnpausedEventAttributes workflow_execution_unpaused_event_attributes = 64;
         WorkflowExecutionTimeSkippingTransitionedEventAttributes workflow_execution_time_skipping_transitioned_event_attributes = 65;
+        WorkflowStreamSubscribedEventAttributes workflow_stream_subscribed_event_attributes = 66;
+        WorkflowStreamRecordsAppendedEventAttributes workflow_stream_records_appended_event_attributes = 67;
     }
 }