[Bridge] Vendored handler progress and told new callers of old records.

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

@@ -0,0 +1,95 @@
+// Nexus service that a Worker implements to take NexusHandler callbacks. The server calls
+// OnComplete for the callback's `operation` and OnProgress for its `progress_operation`. Both
+// operations complete synchronously. The records are backed by the messages in
+// temporal/api/notificationservice/v1/request_response.proto, which hold the full comments.
+package temporal:nexus@1.0.0;
+
+world system {
+  export notification-service;
+}
+
+/// @nexus.service-name "temporal.api.notificationservice.v1.NotificationService"
+/// @nexus.experimental
+interface notification-service {
+  use nexus:temporal-types/model@1.0.0.{
+    failure,
+    payload,
+  };
+
+  /// @nexus.doc "The result of the source operation."
+  /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteRequest.result"
+  variant on-complete-request-result {
+    success(payload),
+    failure(failure),
+  }
+
+  /// @nexus.doc "The completion of the source operation."
+  /// @nexus.experimental
+  /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteRequest"
+  record on-complete-request {
+    /// @nexus.doc "The result of the source operation."
+    %result: on-complete-request-result,
+    /// @nexus.doc "User-supplied data which was added to the source invocation."
+    source-context: option<payload>,
+    /// @nexus.doc "The state of the last progress answer the server accepted, if any."
+    receiver-state: option<list<u8>>,
+  }
+
+  /// @nexus.doc "The answer to a completion."
+  /// @nexus.experimental
+  /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteResponse"
+  record on-complete-response {
+  }
+
+  /// @nexus.doc "The operation that a progress belongs to."
+  /// @nexus.proto "temporal.api.nexus.v1.NexusOperationProgress.operation"
+  variant nexus-operation-progress-operation {
+    scheduled-event-id(s64),
+  }
+
+  /// @nexus.doc "The non-terminal progress of an operation."
+  /// @nexus.experimental
+  /// @nexus.proto "temporal.api.nexus.v1.NexusOperationProgress"
+  record nexus-operation-progress {
+    /// @nexus.doc "The operation this progress belongs to. Not set on a delivery."
+    operation: option<nexus-operation-progress-operation>,
+    /// @nexus.doc "Where the operation's output stands, such as a stream cursor."
+    position: option<string>,
+    /// @nexus.doc "Orders the progress of one operation. The highest counter is kept."
+    counter: option<s64>,
+    /// @nexus.doc "Small details for the caller."
+    metadata: map<string, string>,
+  }
+
+  /// @nexus.doc "A progress of the source operation."
+  /// @nexus.experimental
+  /// @nexus.proto "temporal.api.notificationservice.v1.OnProgressRequest"
+  record on-progress-request {
+    /// @nexus.doc "The progress of the source operation."
+    progress: option<nexus-operation-progress>,
+    /// @nexus.doc "User-supplied data which was added to the source invocation."
+    source-context: option<payload>,
+    /// @nexus.doc "The handler token of a started standalone source operation."
+    operation-token: option<string>,
+    /// @nexus.doc "The state of the last progress answer the server accepted, if any."
+    receiver-state: option<list<u8>>,
+  }
+
+  /// @nexus.doc "The answer to a progress."
+  /// @nexus.experimental
+  /// @nexus.proto "temporal.api.notificationservice.v1.OnProgressResponse"
+  record on-progress-response {
+    /// @nexus.doc "State that the server sends back with the next delivery."
+    receiver-state: option<list<u8>>,
+    /// @nexus.doc "Set when the handler did not reach the progress position."
+    behind: option<bool>,
+  }
+
+  /// @nexus.doc "Receives the completion of the source operation. Called once."
+  /// @nexus.operation name="OnComplete"
+  on-complete: func(request: on-complete-request) -> on-complete-response;
+
+  /// @nexus.doc "Receives the progress of the source operation, any number of times."
+  /// @nexus.operation name="OnProgress"
+  on-progress: func(request: on-progress-request) -> on-progress-response;
+}

crates/protos/protos/api_upstream/openapi/openapiv2.json

@@ -10879,9 +10879,13 @@
           "$ref": "#/definitions/v1Payload",
           "description": "There are restrictions on the maxium payload size a single callback can carry, as well as the\ntotal sum of all source context payloads attached to an execution. See dynamic configuration:\n\"callback.nexusHandler.sourceContext.maxSize\", \"callback.nexusHandler.sourceContext.aggregateMaxSize\".",
           "title": "Arbitrary user-supplied data from the source operation's callsite. (As applicable, not all operations\nsupport attaching context data.)"
+        },
+        "progressOperation": {
+          "type": "string",
+          "description": "Target operation for progress of the source operation, e.g. \"OnProgress\". Empty means the\ncallback takes the completion only."
         }
       },
-      "description": "The targeted Nexus service must be registered within the same namespace as the source operation\nthe callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks\nare strictly caller-side.)\n\nNexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.\nAttempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT\nerror from the server.",
+      "description": "When `progress_operation` is set, the server also invokes that operation of the same service for\nthe source operation's progress. That operation must have the following:\n- Input:  temporal.api.notificationservice.v1.OnProgressRequest\n- Output: temporal.api.notificationservice.v1.OnProgressResponse\nIt may be invoked any number of times before, during or after the single completion. It must\ncomplete synchronously.\n\nThe targeted Nexus service must be registered within the same namespace as the source operation\nthe callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks\nare strictly caller-side.)\n\nNexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.\nAttempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT\nerror from the server.",
       "title": "NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker.\nThe specified Nexus operation must have the following:\n- Input:  temporal.api.notificationservice.v1.OnCompleteRequest\n- Output: temporal.api.notificationservice.v1.OnCompleteResponse"
     },
     "ComputeStatusProviderValidationStatus": {

crates/protos/protos/api_upstream/openapi/openapiv3.yaml

@@ -10935,12 +10935,24 @@ components:
              There are restrictions on the maxium payload size a single callback can carry, as well as the
              total sum of all source context payloads attached to an execution. See dynamic configuration:
              "callback.nexusHandler.sourceContext.maxSize", "callback.nexusHandler.sourceContext.aggregateMaxSize".
+        progressOperation:
+          type: string
+          description: |-
+            Target operation for progress of the source operation, e.g. "OnProgress". Empty means the
+             callback takes the completion only.
       description: |-
         NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker.
          The specified Nexus operation must have the following:
          - Input:  temporal.api.notificationservice.v1.OnCompleteRequest
          - Output: temporal.api.notificationservice.v1.OnCompleteResponse
 
+         When `progress_operation` is set, the server also invokes that operation of the same service for
+         the source operation's progress. That operation must have the following:
+         - Input:  temporal.api.notificationservice.v1.OnProgressRequest
+         - Output: temporal.api.notificationservice.v1.OnProgressResponse
+         It may be invoked any number of times before, during or after the single completion. It must
+         complete synchronously.
+
          The targeted Nexus service must be registered within the same namespace as the source operation
          the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks
          are strictly caller-side.)

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

@@ -207,6 +207,13 @@ message Callback {
     // - Input:  temporal.api.notificationservice.v1.OnCompleteRequest
     // - Output: temporal.api.notificationservice.v1.OnCompleteResponse
     //
+    // When `progress_operation` is set, the server also invokes that operation of the same service for
+    // the source operation's progress. That operation must have the following:
+    // - Input:  temporal.api.notificationservice.v1.OnProgressRequest
+    // - Output: temporal.api.notificationservice.v1.OnProgressResponse
+    // It may be invoked any number of times before, during or after the single completion. It must
+    // complete synchronously.
+    //
     // The targeted Nexus service must be registered within the same namespace as the source operation
     // the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks
     // are strictly caller-side.)
@@ -232,6 +239,10 @@ message Callback {
         // total sum of all source context payloads attached to an execution. See dynamic configuration:
         // "callback.nexusHandler.sourceContext.maxSize", "callback.nexusHandler.sourceContext.aggregateMaxSize".
         temporal.api.common.v1.Payload source_context = 4;
+
+        // Target operation for progress of the source operation, e.g. "OnProgress". Empty means the
+        // callback takes the completion only.
+        string progress_operation = 5;
     }
 
     reserved 1; // For a generic callback mechanism to be added later.

crates/protos/protos/api_upstream/temporal/api/notificationservice/v1/request_response.proto

@@ -32,6 +32,10 @@ message OnCompleteRequest {
 
     // User-supplied data which was added to the source invocation. (As applicable.)
     temporal.api.common.v1.Payload source_context = 3;
+
+    // The `receiver_state` of the last OnProgressResponse that the server accepted for this
+    // callback. Empty when no progress was accepted.
+    bytes receiver_state = 4;
 }
 
 // OnCompleteResponse is the return type of the OnComplete operation.
@@ -48,7 +52,24 @@ message OnProgressRequest {
 
     // User-supplied data which was added to the source invocation. (As applicable.)
     temporal.api.common.v1.Payload source_context = 2;
+
+    // The handler token of the source operation, when the source is a standalone Nexus operation
+    // that has started. Empty otherwise.
+    string operation_token = 3;
+
+    // The `receiver_state` of the last OnProgressResponse that the server accepted for this
+    // callback. Empty when no progress was accepted.
+    bytes receiver_state = 4;
 }
 
 // OnProgressResponse is the return type of the OnProgress operation.
-message OnProgressResponse {}
+message OnProgressResponse {
+    // Opaque state for the handler, such as a read cursor. The server stores it on the callback
+    // and sends it in the next OnProgressRequest and in the OnCompleteRequest. The server refuses
+    // further progress for the callback when the state is larger than its configured limit.
+    bytes receiver_state = 1;
+
+    // Set when the handler did not reach `progress.position`. The server then sends the same
+    // progress again at once, unless newer progress replaced it.
+    bool behind = 2;
+}

crates/protos/src/protos/mod.rs

@@ -3071,6 +3071,22 @@ mod sdk_helpers {
         };
         use anyhow::anyhow;
 
+        mod nexus_handler_callback {
+            use crate::protos::temporal::api::common::v1::callback::NexusHandler;
+            use prost::Message;
+
+            // The server reads the progress operation by its field number, so a renumbered field
+            // would send progress to an operation the Worker never named.
+            #[test]
+            fn the_progress_operation_keeps_its_field_number() {
+                let handler = NexusHandler {
+                    progress_operation: "p".to_string(),
+                    ..Default::default()
+                };
+                assert_eq!(handler.encode_to_vec(), vec![0x2a, 0x01, b'p']);
+            }
+        }
+
         mod stream_record_envelope {
             use crate::protos::temporal::{
                 api::common::v1::Payload,

crates/sdk-core/CHANGELOG.md

@@ -176,6 +176,10 @@ relevant information.
   the Nexus calls, so lang's handler raises it as it is. A mistake in the stream, an empty callback
   URL or request id, and a token that names no stream are `BAD_REQUEST`. An attach for the latest
   run describes the Workflow fresh, so it never lands on a chain that already ended.
+* Experimental: with `notify_on_append`, `AttachCaller` tells the stream's notifier about the
+  newest record after the attach, so a caller that only waits for progress hears of the records
+  already there. An empty stream needs nothing, since every append notifies. The API protos carry the
+  `NexusHandler` progress fields and `nexus/notification-service.wit`.
 * Experimental: a Workflow publishes stream records with `WorkflowOutputStreamCommit` in its
   activation completion. The Worker appends them to its `stream_store` before it reports the
   completion, so readers see them while the Workflow Task runs, and nothing goes to History. The

crates/streams/src/handle.rs

@@ -747,6 +747,11 @@ impl Streams {
         }
     }
 
+    /// The store the handles serve.
+    pub(crate) fn store(&self) -> &dyn StreamStore {
+        self.store.as_ref()
+    }
+
     /// The cursor of a stream's newest record, empty when it holds none.
     pub async fn latest(&self, request: LatestRequest) -> StreamResult<LatestResponse> {
         let stream = address(request.stream.as_ref())?;