Carried the external stream task and replay wake fixes.

.github/workflows/per-pr.yml

@@ -123,7 +123,7 @@ jobs:
 
   integ-tests:
     name: Integ tests
-    timeout-minutes: ${{ github.ref == 'refs/heads/main' && 30 || 25 }}
+    timeout-minutes: ${{ matrix.timeoutMinutes || (github.ref == 'refs/heads/main' && 30 || 25) }}
     strategy:
       fail-fast: false
       matrix:

arch_docs/streaming-poc-docs/decisions/ADR-046-output-capacity-is-logical-and-segmentation-is-shared.md

@@ -22,6 +22,13 @@ per-topic record-count vector to each segment, including empty segments. Replay
 capacity and latency policy, validates the recorded logical manifests, and performs the live number
 of event-loop drains once across both directions.
 
+An input schedule also retains activations with no input observation. If the first subscription
+is created late in a task, earlier empty activations in that task precede its first observed
+segment. This keeps the two schedules aligned when an Activity, timer, or another stream causes
+an intervening activation. A prerelease combined marker with different input and output segment
+counts is rejected: the positions of omitted empty input segments cannot generally be inferred
+from those counts, and dropping output segments would weaken replay verification.
+
 When another publish would exceed the record or logical-byte limit, `publish()` waits, the current
 batch is staged with an output-capacity terminal, and Core forces a replacement Workflow Task. A
 single oversized record or manifest is rejected before unsafe external I/O or marker growth.

crates/client/src/grpc.rs

@@ -955,6 +955,51 @@ proxier! {
             r.extensions_mut().insert(labels);
         }
     );
+    (
+        notify_channel,
+        NotifyChannelRequest,
+        NotifyChannelResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        register_channel_listener,
+        RegisterChannelListenerRequest,
+        RegisterChannelListenerResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        unregister_channel_listener,
+        UnregisterChannelListenerRequest,
+        UnregisterChannelListenerResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        poll_channel,
+        PollChannelRequest,
+        PollChannelResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        describe_channel,
+        DescribeChannelRequest,
+        DescribeChannelResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
     (
         signal_with_start_workflow_execution,
         SignalWithStartWorkflowExecutionRequest,

crates/common/build.rs

@@ -820,6 +820,14 @@ const NOT_VALIDATED_FIELDS: &[&str] = &[
     "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.continued_failure",
     "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.last_completion_result",
     "temporal.api.workflowservice.v1.TerminateWorkflowExecutionRequest.details",
+    // Stream records: the blob limit is a per-event limit, and these bodies never reach an
+    // event. The server bounds the batch by message count (MaxMessagesPerBatch) instead, which
+    // is not a payload size the SDK can mirror.
+    "temporal.api.stream.v1.StreamRecord.body",
+    "temporal.api.stream.v1.StreamRecord.metadata",
+    // Notification metadata: the server bounds it with its own notification size limit, not
+    // the blob limit, so the SDK has nothing to mirror.
+    "temporal.api.notification.v1.Notification.metadata",
     // Dedicated, non-fetchable limits (not blob/memo, not in DescribeNamespace): UserMetadata
     // (nexus-start only); Nexus EndpointSpec.description (maxDescriptionSize; cloud variant cloud-only).
     "temporal.api.sdk.v1.UserMetadata.details",

crates/protos/build.rs

@@ -29,13 +29,15 @@ const SERDE_DERIVE_PREFIXES: &[&str] = &[
     ".temporal.api.namespace",
     ".temporal.api.nexus",
     ".temporal.api.nexusservices",
+    ".temporal.api.notification",
     ".temporal.api.operatorservice",
     ".temporal.api.protocol",
     ".temporal.api.query",
     ".temporal.api.replication",
     ".temporal.api.rules",
     ".temporal.api.schedule",
     ".temporal.api.sdk",
+    ".temporal.api.stream",
     ".temporal.api.taskqueue",
     ".temporal.api.testservice",
     ".temporal.api.update",

crates/protos/protos/api_upstream/nexus/deps/nexus-temporal-types/model.wit

@@ -136,24 +136,15 @@ interface model {
   ///   typescript="common.WorkflowIdReusePolicy"
   ///   dotnet="Temporalio.Api.Enums.V1.WorkflowIdReusePolicy"
   ///   typescript-import="@temporalio/common"
-  enum workflow-id-reuse-policy {
-    allow-duplicate,
-    allow-duplicate-failed-only,
-    reject-duplicate,
-    terminate-if-running,
-  }
+  type workflow-id-reuse-policy = placeholder;
 
   /// @nexus.proto "temporal.api.enums.v1.WorkflowIdConflictPolicy" typescript-import="@temporalio/proto"
   /// @nexus.type
   ///   python="temporalio.common.WorkflowIDConflictPolicy"
   ///   typescript="common.WorkflowIdConflictPolicy"
   ///   dotnet="Temporalio.Api.Enums.V1.WorkflowIdConflictPolicy"
   ///   typescript-import="@temporalio/common"
-  enum workflow-id-conflict-policy {
-    fail,
-    use-existing,
-    terminate-existing,
-  }
+  type workflow-id-conflict-policy = placeholder;
 
   /// @nexus.proto "temporal.api.sdk.v1.UserMetadata" typescript-import="@temporalio/proto"
   /// @nexus.flatten-in-api

crates/protos/protos/api_upstream/nexus/workflow-service.wit

@@ -104,7 +104,8 @@ interface workflow-service {
     started: option<bool>,
     /// @nexus.omit
     signal-link: placeholder,
-    first-execution-run-id: string,
+    /// @nexus.omit
+    first-execution-run-id: placeholder,
   }
 
   /// @nexus.doc

crates/protos/protos/api_upstream/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,59 @@ 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;
+        SubscribeNotificationChannelCommandAttributes
+            subscribe_notification_channel_command_attributes = 22;
+        UnsubscribeNotificationChannelCommandAttributes
+            unsubscribe_notification_channel_command_attributes = 23;
     }
 }
+
+// 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;
+}
+
+// Makes the Workflow a listener of a notification channel for this run. The
+// next notifications on the channel arrive on the scheduled event of a Workflow
+// Task. The subscription ends with the run, and a successor subscribes again.
+message SubscribeNotificationChannelCommandAttributes {
+    // The channel to listen on, as the writers name it.
+    string channel = 1;
+}
+
+// Ends the run's subscription to a notification channel. Notifications already
+// recorded on a scheduled event still reach that Workflow Task; later ones do
+// not. A command naming a channel the run is not subscribed to records its
+// event and changes nothing, so replay matches every command to an event.
+message UnsubscribeNotificationChannelCommandAttributes {
+    // The channel to stop listening on, as the writers name it.
+    string channel = 1;
+}

crates/protos/protos/api_upstream/temporal/api/enums/v1/command_type.proto

@@ -29,4 +29,8 @@ 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;
+    COMMAND_TYPE_SUBSCRIBE_NOTIFICATION_CHANNEL = 21;
+    COMMAND_TYPE_UNSUBSCRIBE_NOTIFICATION_CHANNEL = 22;
 }

crates/protos/protos/api_upstream/temporal/api/enums/v1/event_type.proto

@@ -175,4 +175,19 @@ 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;
+    // A Workflow became a listener of a notification channel for its run.
+    // The notifications themselves ride the WorkflowTaskScheduled event.
+    EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_SUBSCRIBED = 63;
+    // A Workflow stopped listening on a notification channel for its run.
+    // Recorded for every UnsubscribeNotificationChannel command, including one
+    // naming a channel the run was not subscribed to.
+    EVENT_TYPE_WORKFLOW_NOTIFICATION_CHANNEL_UNSUBSCRIBED = 64;
 }

crates/protos/protos/api_upstream/temporal/api/enums/v1/failed_cause.proto

@@ -90,6 +90,19 @@ 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;
+    // A SubscribeNotificationChannel command named an empty or too-long channel, or hit a
+    // subscription or listener limit.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES = 44;
+    // An UnsubscribeNotificationChannel command named an empty or too-long channel.
+    WORKFLOW_TASK_FAILED_CAUSE_BAD_UNSUBSCRIBE_NOTIFICATION_CHANNEL_ATTRIBUTES = 45;
 }
 
 enum StartChildWorkflowExecutionFailedCause {

crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto

@@ -17,6 +17,8 @@ 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/notification/v1/message.proto";
 import "temporal/api/deployment/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
 import "temporal/api/taskqueue/v1/message.proto";
@@ -302,6 +304,10 @@ message WorkflowTaskScheduledEventAttributes {
     google.protobuf.Duration start_to_close_timeout = 2;
     // Starting at 1, how many attempts there have been to complete this task
     int32 attempt = 3;
+    // Notifications for channels this Workflow listens to, folded per channel
+    // since the last task was scheduled. In History so a Workflow may act on
+    // them deterministically and replay sees the same.
+    repeated temporal.api.notification.v1.Notification notifications = 4;
 }
 
 message WorkflowTaskStartedEventAttributes {
@@ -379,6 +385,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 {
@@ -953,6 +966,53 @@ 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 WorkflowNotificationChannelSubscribedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command created this
+    // subscription.
+    int64 workflow_task_completed_event_id = 1;
+    // The channel the Workflow listens on for the rest of this run.
+    string channel = 2;
+}
+
+message WorkflowNotificationChannelUnsubscribedEventAttributes {
+    // The WorkflowTaskCompleted event of the task whose command ended this
+    // subscription.
+    int64 workflow_task_completed_event_id = 1;
+    // The channel the Workflow stopped listening on.
+    string channel = 2;
+    // The WorkflowNotificationChannelSubscribed event that recorded the
+    // subscription this command ended. Zero when the run held no subscription
+    // for the channel.
+    int64 subscribed_event_id = 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;
@@ -1276,6 +1336,12 @@ 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;
+        WorkflowNotificationChannelSubscribedEventAttributes
+            workflow_notification_channel_subscribed_event_attributes = 68;
+        WorkflowNotificationChannelUnsubscribedEventAttributes
+            workflow_notification_channel_unsubscribed_event_attributes = 69;
     }
 }