[Bridge] Attached and detached stream operation callers in Core.

crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto

@@ -6,6 +6,7 @@ option ruby_package = "Temporalio::Internal::Bridge::Api::Streams";
 import "google/protobuf/timestamp.proto";
 import "temporal/api/common/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
+import "temporal/sdk/core/streams/streams.proto";
 
 // The stream service's Nexus operations, served by Core. Lang forwards each call's input in the
 // N0a contract's JSON transfer form, runs its payload codec when Core asks for it, and hands the
@@ -20,6 +21,12 @@ service StreamNexusService {
   rpc NexusRead(NexusCall) returns (NexusStep);
   // Goes on with a call that asked for codec work.
   rpc NexusResume(NexusResumeRequest) returns (NexusStep);
+  // Attaches the caller of a stream-returning Nexus operation's start to the stream's notifier,
+  // and answers with the operation token, or the failure the start answers with. The server then delivers the stream's progress and its
+  // close result to the caller's callback. Idempotent by the start's request id.
+  rpc AttachCaller(AttachCallerRequest) returns (AttachCallerResponse);
+  // Detaches the caller a token's start attached, from the chain it attached to.
+  rpc DetachCaller(DetachCallerRequest) returns (DetachCallerResponse);
 }
 
 // One Nexus operation call.
@@ -91,3 +98,34 @@ message NexusFailure {
   // application error type.
   temporal.api.failure.v1.Failure failure = 4;
 }
+
+message AttachCallerRequest {
+  // The stream the operation's start names. An empty run id attaches to the chain that holds the
+  // Workflow id now.
+  StreamAddress stream = 1;
+  // The start's request id. A retried start attaches once.
+  string request_id = 2;
+  // The start's callback URL and headers. An empty URL is refused, since the caller could never
+  // hear about the stream.
+  string callback_url = 3;
+  map<string, string> callback_headers = 4;
+}
+
+message AttachCallerResponse {
+  // The token the start answers with. It names the stream, the attach and the chain.
+  string operation_token = 1;
+  // Set when the attach failed, as the Nexus calls answer a failure, so lang hands it to its
+  // Nexus handler as it is. The token is then empty.
+  NexusFailure failure = 2;
+}
+
+message DetachCallerRequest {
+  string namespace = 1;
+  // The token the start answered with.
+  string operation_token = 2;
+}
+
+message DetachCallerResponse {
+  // Set when the detach failed, as the Nexus calls answer a failure.
+  NexusFailure failure = 1;
+}

crates/sdk-core/CHANGELOG.md

@@ -167,6 +167,15 @@ relevant information.
   in the same activations. Progress for an operation that is unknown, not started, already
   resolved, or below a counter already delivered is dropped. A task scheduled with progress ends
   a workflow task heartbeat chain, as one with stream wakes does.
+* Experimental: the stream calls `AttachCaller` and `DetachCaller` serve a stream-returning Nexus
+  operation. `AttachCaller` resolves the stream's run chain, mints the operation token and
+  attaches the start's callback to the stream's notifier with the token and start time.
+  `DetachCaller` decodes the token and detaches from the chain the start attached to. The token
+  format is in `crates/streams/nexus/operation_token_vectors.json`, which every SDK's encoder and
+  decoder check. Both answer a failure in the response as a `NexusFailure`, from the same table as
+  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: 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/sdk-core/src/streams.rs

@@ -6,7 +6,7 @@ use std::{collections::BTreeMap, sync::Arc};
 use temporalio_client::Connection;
 use temporalio_common::protos::google::rpc::Status as RpcStatus;
 use temporalio_common::protos::temporal::api::{
-    common::v1::WorkflowExecution,
+    common::v1::{WorkflowExecution, callback},
     enums::v1::StreamOwnerKind,
     stream::v1::{self, StreamReference, WorkflowWakeTarget},
     workflowservice::v1::{
@@ -18,8 +18,9 @@ use temporalio_common::protos::temporal::api::{
     },
 };
 use temporalio_streams::{
-    CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient, OwnerDescription,
-    OwnerError, OwnerId, OwnerStatus, StreamNotification, WorkflowAttachment, proto::ChainId,
+    CallerAttachment, CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient,
+    OwnerDescription, OwnerError, OwnerId, OwnerStatus, StreamNotification, WorkflowAttachment,
+    proto::ChainId,
 };
 pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto};
 use tonic::IntoRequest;
@@ -80,6 +81,15 @@ impl NotifierClient for ConnectionNotifierClient {
             .map_err(notify_error)
     }
 
+    async fn attach_caller(&self, attachment: CallerAttachment) -> Result<(), NotifyError> {
+        self.connection
+            .workflow_service()
+            .attach_stream_callback(caller_attach_request(attachment).into_request())
+            .await
+            .map(|_| ())
+            .map_err(notify_error)
+    }
+
     async fn wait_callback(
         &self,
         chain: ChainId,
@@ -212,6 +222,21 @@ fn attach_request(attachment: WorkflowAttachment) -> AttachStreamCallbackRequest
     }
 }
 
+fn caller_attach_request(attachment: CallerAttachment) -> AttachStreamCallbackRequest {
+    AttachStreamCallbackRequest {
+        namespace: attachment.chain.namespace.clone(),
+        stream_ref: Some(chain_reference(&attachment.chain, attachment.topic)),
+        request_id: attachment.request_id,
+        callback: Some(callback::Nexus {
+            url: attachment.callback_url,
+            header: attachment.callback_headers,
+        }),
+        operation_token: attachment.operation_token,
+        start_time: Some(attachment.start_time.into()),
+        ..Default::default()
+    }
+}
+
 /// One `NotifyStreams` request per namespace, since a call names one.
 fn notify_streams_requests(
     notifications: Vec<StreamNotification>,
@@ -636,6 +661,47 @@ mod tests {
         assert!(request.close_result.is_none());
     }
 
+    fn chain() -> ChainId {
+        ChainId {
+            namespace: "default".to_string(),
+            workflow_id: "owner-1".to_string(),
+            first_run_id: "chain-1".to_string(),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn a_caller_attach_names_the_chains_notifier_and_carries_the_start() {
+        let start_time = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
+        let request = caller_attach_request(CallerAttachment {
+            chain: chain(),
+            topic: "tokens".to_string(),
+            request_id: "attach-1".to_string(),
+            callback_url: "temporal://system".to_string(),
+            callback_headers: [("nexus-callback-token".to_string(), "abc".to_string())].into(),
+            operation_token: "token".to_string(),
+            start_time,
+        });
+        assert_eq!(request.namespace, "default");
+        assert_eq!(
+            request.stream_ref,
+            Some(StreamReference {
+                owner_kind: StreamOwnerKind::Workflow as i32,
+                workflow_id: "owner-1".to_string(),
+                run_id: "chain-1".to_string(),
+                topic: "tokens".to_string(),
+                ..Default::default()
+            })
+        );
+        assert_eq!(request.request_id, "attach-1");
+        let callback = request.callback.unwrap();
+        assert_eq!(callback.url, "temporal://system");
+        assert_eq!(callback.header["nexus-callback-token"], "abc");
+        assert_eq!(request.operation_token, "token");
+        assert_eq!(request.start_time.unwrap().seconds, 1_700_000_000);
+        assert!(request.workflow_target.is_none(), "exactly one target");
+    }
+
     #[test]
     fn a_notification_of_a_workflow_activity_stream_names_its_scheduling_run() {
         let request = notify_request(StreamNotification {
@@ -928,6 +994,8 @@ mod tests {
         stream: proto::StreamAddress,
         workflow_id: String,
         run_id: String,
+        /// The chain the caller's operation token names.
+        token_first_run_id: String,
         // The live tests share the callback listener's address.
         _turn: tokio::sync::MutexGuard<'static, ()>,
     }
@@ -941,9 +1009,8 @@ mod tests {
     /// wake targets.
     async fn live() -> Option<Live> {
         use temporalio_common::protos::temporal::api::{
-            common::v1::WorkflowType,
-            taskqueue::v1::TaskQueue,
-            workflowservice::v1::{AttachStreamCallbackRequest, StartWorkflowExecutionRequest},
+            common::v1::WorkflowType, taskqueue::v1::TaskQueue,
+            workflowservice::v1::StartWorkflowExecutionRequest,
         };
         let Ok(server) = std::env::var("STREAMS_NOTIFIER_SERVER") else {
             eprintln!("set STREAMS_NOTIFIER_SERVER to run the live notifier test");
@@ -984,31 +1051,6 @@ mod tests {
             .into_inner();
         // An earlier live test may have left deliveries behind.
         while deliveries().lock().unwrap().try_recv().is_ok() {}
-        connection
-            .workflow_service()
-            .attach_stream_callback(
-                AttachStreamCallbackRequest {
-                    namespace: "default".to_string(),
-                    stream_ref: Some(StreamReference {
-                        owner_kind: StreamOwnerKind::Workflow as i32,
-                        workflow_id: workflow_id.clone(),
-                        run_id: started.run_id.clone(),
-                        topic: "tokens".to_string(),
-                        ..Default::default()
-                    }),
-                    request_id: "caller".to_string(),
-                    callback: Some(
-                        temporalio_common::protos::temporal::api::common::v1::callback::Nexus {
-                            url: "http://127.0.0.1:7899/callback".to_string(),
-                            ..Default::default()
-                        },
-                    ),
-                    ..Default::default()
-                }
-                .into_request(),
-            )
-            .await
-            .unwrap();
         let service = connect_stream_service(
             proto::StreamStoreConfig {
                 store: Some(proto::stream_store_config::Store::Memory(
@@ -1029,12 +1071,34 @@ mod tests {
             topic: "tokens".to_string(),
             ..Default::default()
         };
+        use prost::Message;
+        let attached = proto::AttachCallerResponse::decode(
+            service
+                .call(
+                    "AttachCaller",
+                    &proto::AttachCallerRequest {
+                        stream: Some(stream.clone()),
+                        request_id: "caller".to_string(),
+                        callback_url: "http://127.0.0.1:7899/callback".to_string(),
+                        ..Default::default()
+                    }
+                    .encode_to_vec(),
+                )
+                .await
+                .unwrap()
+                .as_slice(),
+        )
+        .unwrap();
+        let token =
+            temporalio_streams::OperationToken::decode("default", &attached.operation_token)
+                .unwrap();
         Some(Live {
             connection,
             service,
             stream,
             workflow_id,
             run_id: started.run_id,
+            token_first_run_id: token.chain_first_run_id,
             _turn: turn,
         })
     }
@@ -1113,6 +1177,7 @@ mod tests {
     async fn a_live_notifier_hears_appends_as_progress_and_the_close_as_the_result() {
         use temporalio_common::protos::temporal::api::common::v1::Payload;
         let Some(live) = live().await else { return };
+        assert_eq!(live.token_first_run_id, live.run_id);
         live.append_one().await;
         let (state, body) = live
             .close(proto::CloseRequest {

crates/streams/nexus/operation_token_vectors.json

@@ -0,0 +1,78 @@
+{
+  "comment": "Token v2 of a stream-returning Nexus operation: base64url JSON, no padding, keys sorted, non-ASCII escaped, and no chain_run_id for a ref to the newest chain. Core's encoder and every SDK's encoder write the vectors, and every decoder reads them. Decoders also read decode_only, the null chain_run_id that SDK releases before Core wrote.",
+  "vectors": [
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "owner-1",
+        "topic": "tokens"
+      },
+      "attach": "attach-1",
+      "chain": "chain-1",
+      "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ"
+    },
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "owner-1",
+        "chain_run_id": "run-2",
+        "topic": "tokens"
+      },
+      "attach": "attach-1",
+      "chain": "run-1",
+      "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsiY2hhaW5fcnVuX2lkIjoicnVuLTIiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ"
+    },
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "order/42 \"quoted\"",
+        "topic": "t\\back\nline"
+      },
+      "attach": "req-é",
+      "chain": "chain-ü",
+      "token": "eyJhdHRhY2giOiJyZXEtXHUwMGU5IiwiY2hhaW4iOiJjaGFpbi1cdTAwZmMiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwidG9waWMiOiJ0XFxiYWNrXG5saW5lIiwid29ya2Zsb3dfaWQiOiJvcmRlci80MiBcInF1b3RlZFwiIn0sInYiOjJ9"
+    },
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "日本語-wf",
+        "topic": "emoji-😀"
+      },
+      "attach": "r",
+      "chain": "c",
+      "token": "eyJhdHRhY2giOiJyIiwiY2hhaW4iOiJjIiwicmVmIjp7ImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoiZW1vamktXHVkODNkXHVkZTAwIiwid29ya2Zsb3dfaWQiOiJcdTY1ZTVcdTY3MmNcdThhOWUtd2YifSwidiI6Mn0"
+    },
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "delwf",
+        "topic": "t"
+      },
+      "attach": "a",
+      "chain": "c",
+      "token": "eyJhdHRhY2giOiJhXHUwMDdmIiwiY2hhaW4iOiJjIiwicmVmIjp7ImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoidFx1MDA3ZiIsIndvcmtmbG93X2lkIjoiZGVsXHUwMDdmd2YifSwidiI6Mn0"
+    }
+  ],
+  "decode_only": [
+    {
+      "ref": {
+        "kind": "workflow",
+        "workflow_id": "owner-1",
+        "chain_run_id": null,
+        "topic": "tokens"
+      },
+      "attach": "attach-1",
+      "chain": "chain-1",
+      "token": "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoiY2hhaW4tMSIsInJlZiI6eyJjaGFpbl9ydW5faWQiOm51bGwsImtpbmQiOiJ3b3JrZmxvdyIsInRvcGljIjoidG9rZW5zIiwid29ya2Zsb3dfaWQiOiJvd25lci0xIn0sInYiOjJ9"
+    }
+  ],
+  "refused": [
+    "",
+    "not-a-token",
+    "eyJ2IjogMn0",
+    "eyJ2IjogMSwgImF0dGFjaCI6ICJhdHRhY2gtMSIsICJjaGFpbiI6ICJjaGFpbi0xIiwgInJlZiI6IHsia2luZCI6ICJ3b3JrZmxvdyIsICJ3b3JrZmxvd19pZCI6ICJvd25lci0xIiwgInJ1bl9pZCI6IG51bGwsICJ0b3BpYyI6ICJ0b2tlbnMifX0=",
+    "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwiZXh0cmEiOjEsInJlZiI6eyJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0",
+    "eyJhdHRhY2giOiJhIiwiY2hhaW4iOiJjIiwicmVmIjp7ImN1cnNvciI6IngiLCJraW5kIjoid29ya2Zsb3ciLCJ0b3BpYyI6InQiLCJ3b3JrZmxvd19pZCI6IncifSwidiI6Mn0",
+    "eyJhdHRhY2giOiJhdHRhY2gtMSIsImNoYWluIjoicnVuLTEiLCJyZWYiOnsia2luZCI6IndvcmtmbG93IiwicnVuX2lkIjoicnVuLTIiLCJ0b3BpYyI6InRva2VucyIsIndvcmtmbG93X2lkIjoib3duZXItMSJ9LCJ2IjoyfQ"
+  ]
+}

crates/streams/src/lib.rs

@@ -44,10 +44,11 @@ pub use driver::{ArchiveSink, StreamDriver, StreamDriverOptions};
 pub use error::{StreamError, StreamResult};
 pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions};
 pub use memory::MemoryStore;
+pub use nexus::{OPERATION_TOKEN_VERSION, OperationToken};
 pub use notify::{
-    CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, CLOSE_RETRY_LIMIT, DEFAULT_MAX_NOTIFIERS, FLUSH_LIMIT,
-    Notifier, NotifierClient, NotifierPoll, NotifyError, NotifyingStore, StreamNotification,
-    WorkflowAttachment, progress_counter,
+    CLOSE_RETRY_CAP, CLOSE_RETRY_FIRST, CLOSE_RETRY_LIMIT, CallerAttachment, DEFAULT_MAX_NOTIFIERS,
+    FLUSH_LIMIT, Notifier, NotifierClient, NotifierPoll, NotifyError, NotifyingStore,
+    StreamNotification, WorkflowAttachment, progress_counter,
 };
 pub use notifylog::{ArchiveMark, NotificationEntry, NotificationLog};
 pub use owner::{OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus};

crates/streams/src/nexus/mod.rs

@@ -7,8 +7,12 @@
 //! stateless, so a read goes on at any handler Worker.
 
 mod failure;
+mod operation;
 mod wire;
 
+pub use operation::{OPERATION_TOKEN_VERSION, OperationToken};
+pub(crate) use operation::{attach_caller, detach_caller};
+
 use crate::{
     StreamError, Streams,
     handle::CarriedRead,