[Jetty] Added the standalone Activity owner to streams.

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";
@@ -129,25 +130,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.
@@ -165,6 +178,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 << 11 | min(attempt, 2047)`, 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.
@@ -431,7 +448,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

@@ -539,6 +539,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
@@ -642,6 +643,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,34 @@ 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`.
+* Experimental: a standalone Activity owns streams. A `StreamAddress` of the `ACTIVITY` kind names
+  the Activity id and, optionally, its run. One Activity run is a chain of one run, so every chain
+  rule holds with the run as `first_run_id`. Core checks the owner with
+  `DescribeActivityExecution`. Any terminal status ends the run's streams, and retries, backoff,
+  pause and reset keep them open. A new run of the Activity id ends the earlier run's streams.
+  Cursors hash under the kind `activity`. The Redis keys of an Activity run have the hash tag
+  `{<namespace>:!a:<activity id>:<run id>}`, which no Workflow id makes. `DeleteOwner` takes the
+  Activity kind. `OwnerClient::describe` takes an `OwnerId`, and `OwnerDescription.status` is an
+  `OwnerStatus`. A paused Workflow run now counts as running for a read pinned to it.
+* Experimental: notifier calls name an Activity-owned stream by its kind, its Activity id and its
+  run. Notifications, closes, Workflow reader attaches, detaches and `WaitForNotification` polls
+  all do, and a stream driver turns an Activity log key into that reference. Their cursors hash
+  under the kind `activity`.
+* Experimental: an `ActivityProducer` with `attempt_started_time` writes as the attempt
+  `started_unix_ms << 16 | min(attempt, 65535)`. A retry and a reset attempt start later, so they
+  supersede the earlier attempt, also after a reset sets the Temporal attempt back to 1. An
+  attempt that writes on after a newer one started reads as stale. Without the field the attempt
+  is the Temporal attempt as before. `testdata/streams/activity_attempts.json` in
+  `temporalio-streams` pins the formula for other SDKs.
+* 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
+  holds the new keys.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting

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

@@ -1404,6 +1404,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();