[Jetty] Vendored the Activity owner kind and added it to the stream protos.

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

@@ -16,4 +16,10 @@ enum StreamOwnerKind {
     // failure. If the Workflow's run chain ends with no close and callbacks are attached, the
     // stream notifier closes the stream with a failure within `streamnotifier.ownerCheckInterval`.
     STREAM_OWNER_KIND_WORKFLOW = 1;
+    // A standalone Activity owns the stream. The Activity closes the stream itself, with a result
+    // or a failure. If the Activity run reaches a terminal status with no close and callbacks
+    // are attached, the stream notifier closes the stream within
+    // `streamnotifier.ownerCheckInterval`. A completed run gives no result. A failed, canceled,
+    // terminated or timed-out run gives the run's own failure.
+    STREAM_OWNER_KIND_ACTIVITY = 2;
 }

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

@@ -18,14 +18,18 @@ import "temporal/api/failure/v1/message.proto";
 // Names one stream within a namespace: its owner and its topic.
 message StreamReference {
     temporal.api.enums.v1.StreamOwnerKind owner_kind = 1;
-    // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`.
+    // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. Empty for other kinds.
     string workflow_id = 2;
-    // The id of the first run in the owning Workflow's run chain. The stream notifier is keyed by
-    // it, so one notifier serves the stream across the chain's runs, and a stream that reuses the
-    // Workflow id in a new chain gets a notifier of its own. An empty run id is refused.
+    // For `STREAM_OWNER_KIND_WORKFLOW`, the id of the first run in the owning Workflow's run
+    // chain. The stream notifier is keyed by it, so one notifier serves the stream across the
+    // chain's runs, and a stream that reuses the Workflow id in a new chain gets a notifier of its
+    // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. An empty run id is
+    // refused.
     string run_id = 3;
     // The stream's topic within its owner.
     string topic = 4;
+    // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. Empty for other kinds.
+    string activity_id = 5;
 }
 
 // A callback attached to a stream notifier, as DescribeStreamNotifier reports it.

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

@@ -4,6 +4,7 @@ package coresdk.streams;
 option ruby_package = "Temporalio::Internal::Bridge::Api::Streams";
 
 import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
 import "temporal/api/common/v1/message.proto";
 import "temporal/api/failure/v1/message.proto";
 import "temporal/sdk/streams/v1/message.proto";
@@ -120,25 +121,37 @@ message MemoryStoreConfig {}
 enum StreamOwnerKind {
   STREAM_OWNER_KIND_UNSPECIFIED = 0;
   STREAM_OWNER_KIND_WORKFLOW = 1;
+  // A standalone Activity run. It is a chain of one run, so its streams end with the run.
+  STREAM_OWNER_KIND_ACTIVITY = 2;
 }
 
 // One stream, a topic of an owner in a namespace.
 message StreamAddress {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;
+  // The owning Workflow. Empty for an Activity owner.
   string workflow_id = 3;
   // Pins the stream to the run chain that holds this run. Empty follows the chain that holds the
-  // Workflow id when the call starts.
+  // Workflow id when the call starts. For an Activity owner, the Activity run, and empty names
+  // the newest run of the Activity id when the call starts.
   string run_id = 4;
   string topic = 5;
+  // The owning standalone Activity. Empty for a Workflow owner.
+  string activity_id = 6;
 }
 
 // A Workflow's run chain. Its streams are keyed by its first run, so they follow
-// Continue-as-New, and a new chain on the same Workflow id gets streams of its own.
+// Continue-as-New, and a new chain on the same Workflow id gets streams of its own. A standalone
+// Activity run is a chain of one run, with the run as `first_run_id`.
 message ChainId {
   string namespace = 1;
+  // Empty for an Activity owner.
   string workflow_id = 2;
   string first_run_id = 3;
+  // `UNSPECIFIED` is a Workflow owner, so a chain from before the Activity kind stays valid.
+  StreamOwnerKind owner_kind = 4;
+  // Empty for a Workflow owner.
+  string activity_id = 5;
 }
 
 // A producer that names itself.
@@ -156,6 +169,10 @@ message ActivityProducer {
   string run_id = 2;
   string activity_id = 3;
   int32 attempt = 4;
+  // When the attempt started, as the Activity task gives it. With it, Core writes the attempt as
+  // `started_unix_ms << 16 | min(attempt, 65535)`, so a later attempt and a reset attempt both
+  // supersede an earlier one. Unset writes `attempt` as it is.
+  google.protobuf.Timestamp attempt_started_time = 5;
 }
 
 // One record lang appends.
@@ -422,7 +439,10 @@ message RunStreamDriverResponse {}
 message DeleteOwnerRequest {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;
+  // The Workflow whose streams go, for a Workflow owner.
   string workflow_id = 3;
+  // The standalone Activity whose streams go, across all its runs, for an Activity owner.
+  string activity_id = 4;
 }
 
 message DeleteOwnerResponse {

crates/sdk-core-c-bridge/src/tests/mod.rs

@@ -537,6 +537,7 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() {
             workflow_id: "wf".to_string(),
             run_id: String::new(),
             topic: "out".to_string(),
+            ..Default::default()
         };
 
         let appended = context
@@ -632,6 +633,7 @@ fn test_a_cancelled_stream_read_says_it_only_stopped_reading() {
                 workflow_id: "wf".to_string(),
                 run_id: String::new(),
                 topic: "out".to_string(),
+                ..Default::default()
             }),
             after: "$end".to_string(),
             wait: Some(std::time::Duration::from_secs(30).try_into().unwrap()),

crates/sdk-core/CHANGELOG.md

@@ -144,6 +144,11 @@ relevant information.
   retry would be refused the same way, also when the run closed the topic itself before its task
   reran. Any other store failure still fails the task. Local Activities that a completion with
   stream output schedules start only once that output is in the store.
+* Experimental: the stream protos name a standalone Activity owner. `StreamOwnerKind` gains
+  `ACTIVITY`, `StreamAddress` and `DeleteOwnerRequest` gain `activity_id`, and `ChainId` gains
+  `owner_kind` and `activity_id`. `ActivityProducer` gains `attempt_started_time`. The vendored
+  API's `StreamReference` gains `activity_id` and its owner kind gains `ACTIVITY`. Core still
+  refuses the Activity kind as `UNSUPPORTED`.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting

crates/sdk-core/src/core_tests/external_streams.rs

@@ -1299,6 +1299,7 @@ async fn a_topic_closed_before_the_publish_drops_its_records_and_the_task_goes_o
         namespace: crate::test_help::NAMESPACE.to_string(),
         workflow_id: "fakeid".to_string(),
         first_run_id: String::new(),
+        ..Default::default()
     };
     let appended_at_completion: Arc<Mutex<Vec<usize>>> = Default::default();
     let history = timer_history();

crates/sdk-core/src/streams.rs

@@ -176,6 +176,7 @@ fn chain_reference(chain: &ChainId, topic: String) -> StreamReference {
         workflow_id: chain.workflow_id.clone(),
         run_id: chain.first_run_id.clone(),
         topic,
+        ..Default::default()
     }
 }
 
@@ -434,6 +435,7 @@ mod tests {
                 namespace: "ns".to_string(),
                 workflow_id: "wf".to_string(),
                 first_run_id: "run-1".to_string(),
+                ..Default::default()
             },
             topic: "tokens".to_string(),
             position: "cursor".to_string(),
@@ -448,6 +450,7 @@ mod tests {
                 workflow_id: "wf".to_string(),
                 run_id: "run-1".to_string(),
                 topic: "tokens".to_string(),
+                ..Default::default()
             })
         );
         assert_eq!((request.position.as_str(), request.counter), ("cursor", 7));
@@ -462,6 +465,7 @@ mod tests {
                 namespace: namespace.to_string(),
                 workflow_id: workflow_id.to_string(),
                 first_run_id: "run-1".to_string(),
+                ..Default::default()
             },
             topic: "tokens".to_string(),
             position: format!("cursor-{counter}"),
@@ -506,6 +510,7 @@ mod tests {
                 namespace: "ns".to_string(),
                 workflow_id: "owner".to_string(),
                 first_run_id: "chain-1".to_string(),
+                ..Default::default()
             },
             topic: "tokens".to_string(),
             request_id: "jetty/reader-run/1".to_string(),
@@ -521,6 +526,7 @@ mod tests {
                 workflow_id: "owner".to_string(),
                 run_id: "chain-1".to_string(),
                 topic: "tokens".to_string(),
+                ..Default::default()
             })
         );
         assert_eq!(request.request_id, "jetty/reader-run/1");
@@ -744,6 +750,7 @@ mod tests {
                         workflow_id: workflow_id.clone(),
                         run_id: started.run_id.clone(),
                         topic: "tokens".to_string(),
+                        ..Default::default()
                     }),
                     request_id: "caller".to_string(),
                     callback: Some(
@@ -776,6 +783,7 @@ mod tests {
             workflow_id: workflow_id.clone(),
             run_id: String::new(),
             topic: "tokens".to_string(),
+            ..Default::default()
         };
         Some(Live {
             connection,
@@ -917,6 +925,7 @@ mod tests {
                         workflow_id: live.workflow_id.clone(),
                         run_id: live.run_id.clone(),
                         topic: "tokens".to_string(),
+                        ..Default::default()
                     }),
                 }
                 .into_request(),

crates/sdk-core/src/worker/workflow/stream_output.rs

@@ -173,6 +173,7 @@ impl OutputStore {
             namespace: self.namespace.clone(),
             workflow_id: workflow_id.to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 

crates/streams/src/driver.rs

@@ -472,6 +472,7 @@ impl ChainKey {
             namespace: self.0,
             workflow_id: self.1,
             first_run_id: self.2,
+            ..Default::default()
         }
     }
 }
@@ -580,6 +581,7 @@ mod tests {
             namespace: namespace.to_string(),
             workflow_id: workflow_id.to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 

crates/streams/src/handle.rs

@@ -1023,6 +1023,7 @@ fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId {
         namespace: owner.namespace.clone(),
         workflow_id: owner.workflow_id.clone(),
         first_run_id: first_run_id.to_string(),
+        ..Default::default()
     }
 }
 
@@ -1175,6 +1176,7 @@ mod tests {
             workflow_id: "wf".to_string(),
             run_id: String::new(),
             topic: topic.to_string(),
+            ..Default::default()
         }
     }
 
@@ -1207,6 +1209,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 
@@ -1270,6 +1273,7 @@ mod tests {
             run_id: "run-1".to_string(),
             activity_id: "fetch".to_string(),
             attempt: 2,
+            ..Default::default()
         }));
         setup.streams.append(request).await.unwrap();
         let read = setup
@@ -2466,6 +2470,7 @@ mod tests {
             namespace: "ns".to_string(),
             owner_kind: StreamOwnerKind::Workflow as i32,
             workflow_id: "wf".to_string(),
+            ..Default::default()
         };
         assert!(
             setup

crates/streams/src/memory.rs

@@ -631,6 +631,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 
@@ -691,6 +692,7 @@ mod tests {
                 namespace: "ns".to_string(),
                 owner_kind: StreamOwnerKind::Unspecified as i32,
                 workflow_id: "wf".to_string(),
+                ..Default::default()
             })
             .await
             .unwrap_err();

crates/streams/src/notify.rs

@@ -232,6 +232,7 @@ impl StreamKey {
             namespace: self.namespace.clone(),
             workflow_id: self.workflow_id.clone(),
             first_run_id: self.first_run_id.clone(),
+            ..Default::default()
         }
     }
 }
@@ -934,6 +935,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 

crates/streams/src/reader.rs

@@ -580,6 +580,7 @@ mod tests {
                 namespace: "ns".to_string(),
                 workflow_id: "wf".to_string(),
                 first_run_id: "run-1".to_string(),
+                ..Default::default()
             },
             topic: "out".to_string(),
             stream_hash: stream_hash("ns", "workflow", "wf", "out"),

crates/streams/src/record.rs

@@ -188,6 +188,7 @@ mod tests {
             run_id: "run-1".to_string(),
             activity_id: "fetch".to_string(),
             attempt: 3,
+            ..Default::default()
         };
         assert_eq!(activity_producer_id(&activity), "fetch@run-1");
     }

crates/streams/src/redis/keys.rs

@@ -109,6 +109,7 @@ pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)>
             namespace: unpart(namespace)?,
             workflow_id: unpart(workflow_id)?,
             first_run_id: unpart(first_run_id)?,
+            ..Default::default()
         },
         unpart(topic)?,
     ))
@@ -169,6 +170,7 @@ mod tests {
                     namespace: text(case, "namespace").to_string(),
                     workflow_id: text(case, "workflow_id").to_string(),
                     first_run_id: text(case, "first_run_id").to_string(),
+                    ..Default::default()
                 },
             );
             let topic = text(case, "topic");
@@ -187,6 +189,7 @@ mod tests {
                 namespace: text(case, "namespace").to_string(),
                 workflow_id: text(case, "workflow_id").to_string(),
                 first_run_id: text(case, "first_run_id").to_string(),
+                ..Default::default()
             };
             let prefix = text(case, "prefix");
             let keys = ChainKeys::new(prefix, &chain);
@@ -221,6 +224,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         };
         let pattern = owner_pattern("p", "ns", "wf");
         assert_eq!(pattern, "p:{ns:wf:*");

crates/streams/src/redis/mod.rs

@@ -923,6 +923,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: workflow_id.to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 

crates/streams/src/service.rs

@@ -375,6 +375,7 @@ mod tests {
             workflow_id: "wf".to_string(),
             run_id: String::new(),
             topic: "out".to_string(),
+            ..Default::default()
         }
     }
 
@@ -578,6 +579,7 @@ mod tests {
             namespace: "ns".to_string(),
             owner_kind: StreamOwnerKind::Unspecified as i32,
             workflow_id: "wf".to_string(),
+            ..Default::default()
         };
         let failure = service
             .call("DeleteOwner", &other_owner.encode_to_vec())
@@ -588,6 +590,7 @@ mod tests {
             namespace: "ns".to_string(),
             owner_kind: StreamOwnerKind::Workflow as i32,
             workflow_id: String::new(),
+            ..Default::default()
         };
         let failure = service
             .call("DeleteOwner", &no_id.encode_to_vec())