[Bridge] Served streams that a standalone Activity owns over Nexus.

crates/sdk-core/CHANGELOG.md

@@ -252,6 +252,13 @@ relevant information.
   Activity id, keeps version 2, and leaves `activity_id` out of a Workflow's ref, so a Workflow's
   token keeps its bytes. `operation_token_vectors.json` has Activity tokens. `CodecWork` gains
   `owner_kind` and `activity_id`. The Nexus calls still refuse an Activity ref.
+* Experimental: the stream service's Nexus calls serve streams that a standalone Activity owns.
+  `append`, `read`, `read_range`, `latest_only`, the floor holder, `AttachCaller` with
+  `wake_after_counter`, `DetachCaller` and the read-first wake all take an Activity ref. A read's
+  `wake_token` must name the read's owner kind, Workflow id, Activity id and topic, so a token of
+  one Activity never wakes another Activity's reader on the same topic. `CodecWork` names the
+  Activity, so lang's codec context is the namespace and the Activity id. A ref of kind
+  `activity` with a Workflow id, or of kind `workflow` with an Activity id, is a bad request.
 * Experimental: the Activity attempt key is `started_unix_ms << 11 | min(attempt, 2047)`, so it
   fits 53 bits and a JSON number holds it exactly. The Nexus contract's models and JavaScript
   refused the earlier 16-bit form. A start time from May 2109 on is refused. The vector file

crates/streams/src/handle.rs

@@ -478,8 +478,8 @@ impl Streams {
             Ok(described) if described.first_run_id != first_run_id => {
                 return Err(StreamError::cursor(format!(
                     "the read state names run {first_run_id:?}, which doesn't start a run chain \
-                     of Workflow {:?}",
-                    stream.workflow_id
+                     of {}",
+                    key.label()
                 )));
             }
             // A chain Temporal no longer holds leaves nothing another chain's reader could

crates/streams/src/nexus/mod.rs

@@ -272,14 +272,7 @@ impl NexusCalls {
         if finish {
             return self.finish(work, Vec::new()).await;
         }
-        self.ask_codec(
-            work,
-            bytes,
-            true,
-            &call.namespace,
-            &input.stream.workflow_id,
-            parsed,
-        )
+        self.ask_codec(work, bytes, true, &call.namespace, &input.stream, parsed)
     }
 
     /// Serves the `read` operation: the records after the caller's cursor, waiting for some.
@@ -315,8 +308,12 @@ impl NexusCalls {
         let floor_holder = input.floor_holder.unwrap_or_default();
         let wake = match &input.wake_token {
             Some(token) => match OperationToken::decode(&call.namespace, token) {
+                // A token of another owner with the same topic must not wake this reader, and
+                // every Activity's ref has the same empty Workflow id.
                 Ok(token)
-                    if token.stream.workflow_id == address.workflow_id
+                    if token.stream.owner_kind == address.owner_kind
+                        && token.stream.workflow_id == address.workflow_id
+                        && token.stream.activity_id == address.activity_id
                         && token.stream.topic == address.topic =>
                 {
                     Some(token)
@@ -453,14 +450,7 @@ impl NexusCalls {
         if payloads.is_empty() {
             return self.finish(work, Vec::new()).await;
         }
-        self.ask_codec(
-            work,
-            bytes,
-            false,
-            &call.namespace,
-            &input.stream.workflow_id,
-            payloads,
-        )
+        self.ask_codec(work, bytes, false, &call.namespace, &input.stream, payloads)
     }
 
     /// Asks the notifier to wake the operation callback that `token` names once the stream moves
@@ -487,12 +477,7 @@ impl NexusCalls {
         }
         match client
             .wait_callback(
-                crate::proto::ChainId {
-                    namespace: address.namespace.clone(),
-                    workflow_id: address.workflow_id.clone(),
-                    first_run_id: token.chain_first_run_id.clone(),
-                    ..Default::default()
-                },
+                crate::handle::chain_of(address, &token.chain_first_run_id),
                 address.topic.clone(),
                 token.attach_request_id.clone(),
                 read.response.wake_after,
@@ -596,14 +581,7 @@ impl NexusCalls {
         if payloads.is_empty() {
             return self.finish(work, Vec::new()).await;
         }
-        self.ask_codec(
-            work,
-            bytes,
-            false,
-            &call.namespace,
-            &input.stream.workflow_id,
-            payloads,
-        )
+        self.ask_codec(work, bytes, false, &call.namespace, &input.stream, payloads)
     }
 
     /// `address` without its run, and a state that names the chain the run picked. A state the
@@ -660,7 +638,7 @@ impl NexusCalls {
         bytes: usize,
         encode: bool,
         namespace: &str,
-        workflow_id: &str,
+        stream: &StreamRef,
         payloads: Vec<Payload>,
     ) -> NexusStep {
         let continuation = self.next.fetch_add(1, Ordering::Relaxed);
@@ -696,10 +674,13 @@ impl NexusCalls {
             step: Some(nexus_step::Step::Codec(CodecWork {
                 encode,
                 namespace: namespace.to_string(),
-                workflow_id: workflow_id.to_string(),
+                workflow_id: stream.workflow_id.clone(),
                 payloads,
                 continuation,
-                ..Default::default()
+                // The address check already refused any other kind.
+                owner_kind: operation::owner_kind(&stream.kind).unwrap_or(StreamOwnerKind::Workflow)
+                    as i32,
+                activity_id: stream.activity_id.clone().unwrap_or_default(),
             })),
         }
     }
@@ -1144,31 +1125,45 @@ fn check_holder(floor_holder: &str, after: &str) -> Result<(), Box<NexusFailure>
 }
 
 fn address(namespace: &str, stream: &StreamRef) -> Result<StreamAddress, Box<NexusFailure>> {
-    if stream.kind != "workflow" {
+    let Some(owner_kind) = operation::owner_kind(&stream.kind) else {
         return Err(Box::new(failure::of_stream(&StreamError::unsupported(
             format!(
-                "this release opens streams of Workflows only, not of kind {}",
+                "this release opens streams of Workflows and standalone Activities only, not of \
+                 kind {}",
                 failure::named(&stream.kind)
             ),
         ))));
-    }
+    };
     if let Err(reason) = temporalio_common::streams::check_topic(&stream.topic) {
         return Err(Box::new(failure::input_mistake(failure::without_values(
             &reason,
         ))));
     }
-    if stream.workflow_id.is_empty() {
-        return Err(Box::new(failure::input_mistake(
-            "a stream needs its owner's Workflow id",
-        )));
+    let activity_id = stream.activity_id.clone().unwrap_or_default();
+    let mistake = match owner_kind {
+        StreamOwnerKind::Activity if activity_id.is_empty() => {
+            Some("a stream that an Activity owns needs its owner's Activity id")
+        }
+        StreamOwnerKind::Activity if !stream.workflow_id.is_empty() => {
+            Some("a stream that an Activity owns names an empty Workflow id")
+        }
+        StreamOwnerKind::Activity => None,
+        _ if stream.workflow_id.is_empty() => Some("a stream needs its owner's Workflow id"),
+        _ if stream.activity_id.is_some() => {
+            Some("a stream that a Workflow owns names no Activity id")
+        }
+        _ => None,
+    };
+    if let Some(mistake) = mistake {
+        return Err(Box::new(failure::input_mistake(mistake)));
     }
     Ok(StreamAddress {
         namespace: namespace.to_string(),
-        owner_kind: StreamOwnerKind::Workflow as i32,
+        owner_kind: owner_kind as i32,
         workflow_id: stream.workflow_id.clone(),
         run_id: stream.chain_run_id.clone().unwrap_or_default(),
         topic: stream.topic.clone(),
-        ..Default::default()
+        activity_id,
     })
 }
 

crates/streams/src/nexus/operation.rs

@@ -200,16 +200,26 @@ async fn detach(
 /// crosses as the caller's input mistake.
 fn checked(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> {
     let stream = stream.ok_or_else(|| bad_input("an attach needs its stream"))?;
-    if stream.owner_kind != StreamOwnerKind::Workflow as i32 {
-        return Err(StreamError::unsupported(format!(
-            "this release opens streams of Workflows only, not owner kind {}",
-            stream.owner_kind
+    let (owner_id, other_id, kind) = match StreamOwnerKind::try_from(stream.owner_kind) {
+        Ok(StreamOwnerKind::Workflow) => (&stream.workflow_id, &stream.activity_id, "Workflow"),
+        Ok(StreamOwnerKind::Activity) => (&stream.activity_id, &stream.workflow_id, "Activity"),
+        _ => {
+            return Err(StreamError::unsupported(format!(
+                "this release opens streams of Workflows and standalone Activities only, not \
+                 owner kind {}",
+                stream.owner_kind
+            )));
+        }
+    };
+    if stream.namespace.is_empty() || owner_id.is_empty() {
+        return Err(bad_input(&format!(
+            "a stream needs its namespace and its owner's {kind} id"
         )));
     }
-    if stream.namespace.is_empty() || stream.workflow_id.is_empty() {
-        return Err(bad_input(
-            "a stream needs its namespace and its owner's Workflow id",
-        ));
+    if !other_id.is_empty() {
+        return Err(bad_input(&format!(
+            "a stream that a {kind} owns names only its {kind} id"
+        )));
     }
     // The topic is the caller's own value, so the message names the rule and not the topic.
     temporalio_common::streams::check_topic(&stream.topic)
@@ -737,6 +747,49 @@ mod tests {
         assert!(raised.retryable, "{raised:?}");
     }
 
+    #[tokio::test]
+    async fn an_activity_stream_start_attaches_waits_and_detaches_on_the_activity_run() {
+        let setup = setup();
+        let activity = StreamAddress {
+            owner_kind: StreamOwnerKind::Activity as i32,
+            workflow_id: String::new(),
+            activity_id: "tool-1".to_string(),
+            ..stream("")
+        };
+        let start = AttachCallerRequest {
+            stream: Some(activity.clone()),
+            wake_after_counter: Some(0),
+            ..attach_request("temporal://system")
+        };
+        let token = attach(&setup.service, start).await.unwrap();
+        let run = ChainId {
+            namespace: "default".to_string(),
+            first_run_id: "chain-1".to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: "tool-1".to_string(),
+            ..Default::default()
+        };
+        assert_eq!(setup.notifier.attached.lock().unwrap()[0].chain, run);
+        assert_eq!(setup.notifier.waits.lock().unwrap()[0].0, run);
+        let decoded = OperationToken::decode("default", &token).unwrap();
+        assert_eq!(decoded.stream, activity);
+        assert_eq!(decoded.chain(), run);
+        detach(&setup.service, &token).await.unwrap();
+        assert_eq!(setup.notifier.detached.lock().unwrap()[0].0, run);
+        // An Activity's address names no Workflow id.
+        let refused = AttachCallerRequest {
+            stream: Some(StreamAddress {
+                workflow_id: "owner-1".to_string(),
+                ..activity
+            }),
+            ..attach_request("temporal://system")
+        };
+        assert_eq!(
+            attach(&setup.service, refused).await.unwrap_err().kind,
+            "BAD_REQUEST"
+        );
+    }
+
     #[tokio::test]
     async fn a_start_attaches_an_idle_callback_and_announces_nothing() {
         let setup = setup();