diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto
index 4a079844ce..6c2dfa80c5 100644
--- a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto
+++ b/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;
 }
diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto
index e92037ad87..c0a76d5989 100644
--- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto
+++ b/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.
diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
index 1a06a64791..6f43676b90 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/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 {
diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs
index b50e74beb0..890aa4a771 100644
--- a/crates/sdk-core-c-bridge/src/tests/mod.rs
+++ b/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()),
diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md
index a78ed897a8..6de48a00ec 100644
--- a/crates/sdk-core/CHANGELOG.md
+++ b/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
diff --git a/crates/sdk-core/src/core_tests/external_streams.rs b/crates/sdk-core/src/core_tests/external_streams.rs
index 642998563a..1f216fecda 100644
--- a/crates/sdk-core/src/core_tests/external_streams.rs
+++ b/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();
diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs
index 341fb07c1c..80c98a7a6b 100644
--- a/crates/sdk-core/src/streams.rs
+++ b/crates/sdk-core/src/streams.rs
@@ -10,14 +10,15 @@ use temporalio_common::protos::temporal::api::{
     enums::v1::StreamOwnerKind,
     stream::v1::{self, StreamReference, WorkflowWakeTarget},
     workflowservice::v1::{
-        AttachStreamCallbackRequest, DescribeWorkflowExecutionRequest,
+        AttachStreamCallbackRequest, DescribeActivityExecutionRequest,
+        DescribeActivityExecutionResponse, DescribeWorkflowExecutionRequest,
         DescribeWorkflowExecutionResponse, DetachStreamCallbackRequest, NotifyStreamRequest,
         NotifyStreamsRequest, PollStreamNotifierRequest, PollStreamNotifierResponse,
     },
 };
 use temporalio_streams::{
     CloseOutcome, NotifierClient, NotifierPoll, NotifyError, OwnerClient, OwnerDescription,
-    OwnerError, StreamNotification, WorkflowAttachment, proto::ChainId,
+    OwnerError, OwnerId, OwnerStatus, StreamNotification, WorkflowAttachment, proto::ChainId,
 };
 pub use temporalio_streams::{StreamError, StreamResult, StreamService, StreamStore, proto};
 use tonic::IntoRequest;
@@ -169,13 +170,23 @@ fn poll_answer(response: PollStreamNotifierResponse) -> NotifierPoll {
 
 /// The notifier of `chain`'s stream on `topic`. The run chain's first run keys the notifier, so one
 /// notifier serves the stream across Continue-as-New and a new chain on the same Workflow id gets
-/// its own.
+/// its own. An Activity run is its own chain, so its notifier is keyed by the run.
 fn chain_reference(chain: &ChainId, topic: String) -> StreamReference {
+    if chain.owner_kind == proto::StreamOwnerKind::Activity as i32 {
+        return StreamReference {
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: chain.activity_id.clone(),
+            run_id: chain.first_run_id.clone(),
+            topic,
+            ..Default::default()
+        };
+    }
     StreamReference {
         owner_kind: StreamOwnerKind::Workflow as i32,
         workflow_id: chain.workflow_id.clone(),
         run_id: chain.first_run_id.clone(),
         topic,
+        ..Default::default()
     }
 }
 
@@ -281,25 +292,46 @@ impl OwnerClient for ConnectionOwnerClient {
     async fn describe(
         &self,
         namespace: &str,
-        workflow_id: &str,
+        owner: OwnerId<'_>,
         run_id: &str,
     ) -> Result<OwnerDescription, OwnerError> {
-        let response = self
-            .connection
-            .workflow_service()
-            .describe_workflow_execution(
-                DescribeWorkflowExecutionRequest {
-                    namespace: namespace.to_string(),
-                    execution: Some(WorkflowExecution {
-                        workflow_id: workflow_id.to_string(),
-                        run_id: run_id.to_string(),
-                    }),
-                }
-                .into_request(),
-            )
-            .await
-            .map_err(owner_error)?;
-        owner_description(response.into_inner())
+        match owner {
+            OwnerId::Workflow(workflow_id) => {
+                let response = self
+                    .connection
+                    .workflow_service()
+                    .describe_workflow_execution(
+                        DescribeWorkflowExecutionRequest {
+                            namespace: namespace.to_string(),
+                            execution: Some(WorkflowExecution {
+                                workflow_id: workflow_id.to_string(),
+                                run_id: run_id.to_string(),
+                            }),
+                        }
+                        .into_request(),
+                    )
+                    .await
+                    .map_err(owner_error)?;
+                owner_description(response.into_inner())
+            }
+            OwnerId::Activity(activity_id) => {
+                let response = self
+                    .connection
+                    .workflow_service()
+                    .describe_activity_execution(
+                        DescribeActivityExecutionRequest {
+                            namespace: namespace.to_string(),
+                            activity_id: activity_id.to_string(),
+                            run_id: run_id.to_string(),
+                            ..Default::default()
+                        }
+                        .into_request(),
+                    )
+                    .await
+                    .map_err(owner_error)?;
+                activity_owner_description(response.into_inner())
+            }
+        }
     }
 }
 
@@ -341,12 +373,36 @@ fn owner_description(
         info.first_run_id.clone()
     };
     Ok(OwnerDescription {
-        status: info.status(),
+        status: OwnerStatus::of_workflow(info.status()),
         run_id,
         first_run_id,
     })
 }
 
+/// A standalone Activity run is a chain of one run, so it is its own first run.
+fn activity_owner_description(
+    response: DescribeActivityExecutionResponse,
+) -> Result<OwnerDescription, OwnerError> {
+    let info = response
+        .info
+        .ok_or_else(|| OwnerError::Failed("the server described no Activity".to_string()))?;
+    let run_id = if response.run_id.is_empty() {
+        info.run_id.clone()
+    } else {
+        response.run_id
+    };
+    if run_id.is_empty() {
+        return Err(OwnerError::Failed(
+            "the server described an Activity with no run id".to_string(),
+        ));
+    }
+    Ok(OwnerDescription {
+        status: OwnerStatus::of_activity(info.status()),
+        first_run_id: run_id.clone(),
+        run_id,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -376,11 +432,54 @@ mod tests {
             OwnerDescription {
                 run_id: "run-2".to_string(),
                 first_run_id: "run-1".to_string(),
-                status: WorkflowExecutionStatus::ContinuedAsNew,
+                status: OwnerStatus::ContinuedAsNew,
             }
         );
     }
 
+    #[test]
+    fn an_activity_run_is_its_own_chain_and_any_terminal_status_ends_it() {
+        use temporalio_common::protos::temporal::api::{
+            activity::v1::ActivityExecutionInfo, enums::v1::ActivityExecutionStatus,
+        };
+        let described = |status: ActivityExecutionStatus| DescribeActivityExecutionResponse {
+            run_id: "act-run".to_string(),
+            info: Some(ActivityExecutionInfo {
+                activity_id: "gen".to_string(),
+                run_id: "act-run".to_string(),
+                status: status as i32,
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert_eq!(
+            activity_owner_description(described(ActivityExecutionStatus::Paused)).unwrap(),
+            OwnerDescription {
+                run_id: "act-run".to_string(),
+                first_run_id: "act-run".to_string(),
+                status: OwnerStatus::Running,
+            }
+        );
+        for status in [
+            ActivityExecutionStatus::Completed,
+            ActivityExecutionStatus::Failed,
+            ActivityExecutionStatus::Canceled,
+            ActivityExecutionStatus::Terminated,
+            ActivityExecutionStatus::TimedOut,
+        ] {
+            assert!(
+                activity_owner_description(described(status))
+                    .unwrap()
+                    .chain_ended(),
+                "{status:?}"
+            );
+        }
+        assert!(matches!(
+            activity_owner_description(DescribeActivityExecutionResponse::default()),
+            Err(OwnerError::Failed(_))
+        ));
+    }
+
     #[test]
     fn a_run_without_a_first_run_id_starts_its_chain() {
         assert_eq!(
@@ -434,6 +533,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 +548,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));
@@ -455,6 +556,34 @@ mod tests {
         assert!(request.close_result.is_none());
     }
 
+    #[test]
+    fn a_notification_of_an_activity_stream_names_the_activity_run() {
+        let request = notify_request(StreamNotification {
+            chain: proto::ChainId {
+                namespace: "ns".to_string(),
+                first_run_id: "act-run".to_string(),
+                owner_kind: proto::StreamOwnerKind::Activity as i32,
+                activity_id: "gen".to_string(),
+                ..Default::default()
+            },
+            topic: "tokens".to_string(),
+            position: "cursor".to_string(),
+            counter: 7,
+            close: Some(CloseOutcome::Failure(Default::default())),
+        });
+        assert_eq!(
+            request.stream_ref,
+            Some(StreamReference {
+                owner_kind: StreamOwnerKind::Activity as i32,
+                workflow_id: String::new(),
+                run_id: "act-run".to_string(),
+                topic: "tokens".to_string(),
+                activity_id: "gen".to_string(),
+            })
+        );
+        assert!(request.close && request.close_failure.is_some());
+    }
+
     #[test]
     fn a_batch_of_notifications_goes_out_once_per_namespace() {
         let notification = |namespace: &str, workflow_id: &str, counter: i64| StreamNotification {
@@ -462,6 +591,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 +636,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 +652,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 +876,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 +909,7 @@ mod tests {
             workflow_id: workflow_id.clone(),
             run_id: String::new(),
             topic: "tokens".to_string(),
+            ..Default::default()
         };
         Some(Live {
             connection,
@@ -917,6 +1051,7 @@ mod tests {
                         workflow_id: live.workflow_id.clone(),
                         run_id: live.run_id.clone(),
                         topic: "tokens".to_string(),
+                        ..Default::default()
                     }),
                 }
                 .into_request(),
diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs
index 0ab774f560..21c49351ea 100644
--- a/crates/sdk-core/src/worker/workflow/stream_output.rs
+++ b/crates/sdk-core/src/worker/workflow/stream_output.rs
@@ -181,6 +181,7 @@ impl OutputStore {
             namespace: self.namespace.clone(),
             workflow_id: workflow_id.to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 
diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md
index 1acf929ca3..2560d911d5 100644
--- a/crates/streams/docs/redis.md
+++ b/crates/streams/docs/redis.md
@@ -71,9 +71,15 @@ slides the expiry of the log and its meta, so a stream dies a retention after it
 and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as
 the tombstone that tells a log that expired from one that never existed.
 
-**Deletes.** Deleting an owner's streams scans every primary for the Workflow id's keys. `SCAN`
-makes no promise about keys that move to another node while it runs, so a delete during a
-cluster reshard can miss some. Run it again once the reshard is done.
+**Activity owners.** A standalone Activity run is a chain of one run. Its keys have the hash tag
+`{<namespace>:!a:<activity id>:<run id>}`, and every other part of the layout is the same. Each
+part escapes `!`, so no Workflow id makes that tag, and a Workflow's delete pattern never matches
+an Activity's keys. The tag starts with the namespace, so the documented ACL covers it.
+
+**Deletes.** Deleting an owner's streams scans every primary for the owner id's keys, across all
+the Workflow's chains or all the Activity's runs. `SCAN` makes no promise about keys that move to
+another node while it runs, so a delete during a cluster reshard can miss some. Run it again once
+the reshard is done.
 
 **Server.** Redis 7.0 or later is required, and the store refuses an older server when it
 connects. At the same moment it reads `maxmemory-policy` on every primary and logs a warning
diff --git a/crates/streams/src/close.rs b/crates/streams/src/close.rs
index 379c525ecf..276af80c8e 100644
--- a/crates/streams/src/close.rs
+++ b/crates/streams/src/close.rs
@@ -140,6 +140,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 
diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs
index 56f61c1a3d..163aa6d20e 100644
--- a/crates/streams/src/driver.rs
+++ b/crates/streams/src/driver.rs
@@ -9,8 +9,8 @@
 
 use crate::{
     NotificationEntry, NotificationLog, NotifierClient, NotifyError, StreamError,
-    StreamNotification, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor,
-    progress_counter, proto::ChainId, stream_hash,
+    StreamNotification, StreamResult, StreamStore, chain_owner, mint_cursor, progress_counter,
+    proto::ChainId, stream_hash,
 };
 use std::{
     collections::{BTreeMap, HashMap},
@@ -412,6 +412,7 @@ impl StreamDriver {
             let Some(counter) = progress_counter(&entry.position) else {
                 tracing::warn!(
                     workflow_id = %entry.chain.workflow_id,
+                    activity_id = %entry.chain.activity_id,
                     topic = %entry.topic,
                     position = %entry.position,
                     "No notifier counter comes from this position, so it isn't notified"
@@ -435,12 +436,8 @@ impl StreamDriver {
             .into_iter()
             .map(|((chain, topic), (counter, position, ids))| {
                 let chain = chain.into_chain();
-                let hash = stream_hash(
-                    &chain.namespace,
-                    WORKFLOW_OWNER_KIND,
-                    &chain.workflow_id,
-                    &topic,
-                );
+                let (kind, id) = chain_owner(&chain);
+                let hash = stream_hash(&chain.namespace, kind, id, &topic);
                 let notification = StreamNotification {
                     position: mint_cursor(self.store.name(), &hash, &position),
                     chain,
@@ -454,9 +451,9 @@ impl StreamDriver {
     }
 }
 
-/// A chain as a map key.
+/// A chain as a map key: its namespace, Workflow id, first run, owner kind and Activity id.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
-struct ChainKey(String, String, String);
+struct ChainKey(String, String, String, i32, String);
 
 impl ChainKey {
     fn of(chain: &ChainId) -> Self {
@@ -464,6 +461,8 @@ impl ChainKey {
             chain.namespace.clone(),
             chain.workflow_id.clone(),
             chain.first_run_id.clone(),
+            chain.owner_kind,
+            chain.activity_id.clone(),
         )
     }
 
@@ -472,6 +471,8 @@ impl ChainKey {
             namespace: self.0,
             workflow_id: self.1,
             first_run_id: self.2,
+            owner_kind: self.3,
+            activity_id: self.4,
         }
     }
 }
@@ -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()
         }
     }
 
@@ -662,6 +664,46 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn an_activity_chains_entries_notify_the_activitys_stream() {
+        let store = Arc::new(MemoryStore::with_notification_log());
+        let client = Arc::new(Recorder::default());
+        let activity = ChainId {
+            namespace: "ns".to_string(),
+            first_run_id: "run-1".to_string(),
+            owner_kind: crate::proto::StreamOwnerKind::Activity as i32,
+            activity_id: "a".to_string(),
+            ..Default::default()
+        };
+        // A Workflow with the same id and run writes another stream.
+        append(&store, "a", "out", 1).await;
+        let landed = store
+            .append(StoreAppendRequest {
+                chain: Some(activity.clone()),
+                topic: "out".to_string(),
+                producer_id: "p".to_string(),
+                attempt: 1,
+                sequence: 1,
+                digest: vec![1],
+                records: vec![b"record".to_vec(), b"record".to_vec()],
+            })
+            .await
+            .unwrap();
+        driver(&store, &client, "d1").drain(0).await.unwrap();
+        let calls = client.calls.lock().unwrap().clone();
+        assert_eq!(calls[0].len(), 2, "two streams, {calls:?}");
+        let sent = calls[0].iter().find(|n| n.chain == activity).unwrap();
+        assert_eq!(sent.counter, 2);
+        assert_eq!(
+            sent.position,
+            mint_cursor(
+                "memory",
+                &stream_hash("ns", crate::ACTIVITY_OWNER_KIND, "a", "out"),
+                &landed.last_position
+            )
+        );
+    }
+
     #[tokio::test]
     async fn a_failed_call_leaves_the_entries_for_the_next_drain() {
         let store = Arc::new(MemoryStore::with_notification_log());
diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs
index 6a79cf6a69..cf8252ff3f 100644
--- a/crates/streams/src/handle.rs
+++ b/crates/streams/src/handle.rs
@@ -5,9 +5,9 @@
 
 use crate::{
     BEGINNING, END, MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore,
-    WORKFLOW_OWNER_KIND, activity_producer_id, append_batch, check_append_digest,
+    activity_attempt, activity_producer_id, append_batch, chain_owner, check_append_digest,
     check_store_limits, close_topic, cursor_position, mint_cursor,
-    owner::{OwnerClient, OwnerDescription, OwnerError},
+    owner::{OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus},
     proto::{
         ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse,
         DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer,
@@ -25,7 +25,6 @@ use std::{
     sync::{Arc, Mutex},
     time::Duration,
 };
-use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus;
 use tokio::time::Instant;
 
 /// The longest `floor_holder` a read takes.
@@ -83,10 +82,32 @@ impl StreamsOptions {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct OwnerKey {
     namespace: String,
-    workflow_id: String,
+    activity: bool,
+    /// The Workflow id or the Activity id.
+    id: String,
     run_id: String,
 }
 
+impl OwnerKey {
+    fn owner(&self) -> OwnerId<'_> {
+        if self.activity {
+            OwnerId::Activity(&self.id)
+        } else {
+            OwnerId::Workflow(&self.id)
+        }
+    }
+
+    /// The owner as error texts name it.
+    fn label(&self) -> String {
+        let kind = if self.activity {
+            "Activity"
+        } else {
+            "Workflow"
+        };
+        format!("{kind} {:?}", self.id)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct ProducerKey {
     owner: OwnerKey,
@@ -200,10 +221,11 @@ impl Streams {
     /// Appends one batch for one producer attempt.
     ///
     /// The first batch for an owner, and the first after [StreamsOptions::owner_recheck], asks
-    /// whether the owner's chain ended. All producers of the owner share that check. If it did,
-    /// or Temporal no longer holds the owner, or a new chain took the Workflow id, the producer
-    /// closes the chain, or its own topic on a store without close flags, and the batch is
-    /// refused as closed.
+    /// whether the owner's chain ended. All producers of the owner share that check. If it did, or
+    /// Temporal no longer holds the owner, or a new chain took the owner id, the producer closes
+    /// the chain, or its own topic on a store without close flags, and the batch is refused as
+    /// closed. A standalone Activity run is a chain of one run, so a new run of the Activity id
+    /// ends the earlier run's streams.
     pub async fn append(&self, request: AppendRequest) -> StreamResult<AppendResponse> {
         let stream = address(request.stream.as_ref())?;
         let (producer_id, attempt) = match &request.producer {
@@ -600,13 +622,11 @@ impl Streams {
     async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult<bool> {
         let key = owner_key(stream);
         match self.describe(&key, &key.run_id).await {
-            // The chain came from this Workflow, so History has since dropped it, or a failover
+            // The chain came from this owner, so Temporal has since dropped it, or a failover
             // hides it for a while.
             Err(OwnerError::NotFound(_)) => Ok(self.gone_for_good(chain)),
             Err(error) => Err(describe_failed(&error)),
-            Ok(described) if !key.run_id.is_empty() => {
-                Ok(described.status != WorkflowExecutionStatus::Running)
-            }
+            Ok(described) if !key.run_id.is_empty() => Ok(described.status != OwnerStatus::Running),
             Ok(latest) => Ok(latest.chain_ended() || latest.first_run_id != chain.first_run_id),
         }
     }
@@ -720,22 +740,26 @@ impl Streams {
         &self,
         request: DeleteOwnerRequest,
     ) -> StreamResult<DeleteOwnerResponse> {
-        if request.owner_kind != StreamOwnerKind::Workflow as i32 {
-            return Err(StreamError::unsupported(format!(
-                "this release keeps streams of Workflows only, not owner kind {}",
-                request.owner_kind
-            )));
+        let (id, other) = match StreamOwnerKind::try_from(request.owner_kind) {
+            Ok(StreamOwnerKind::Workflow) => (&request.workflow_id, &request.activity_id),
+            Ok(StreamOwnerKind::Activity) => (&request.activity_id, &request.workflow_id),
+            _ => return Err(unsupported_kind(request.owner_kind)),
+        };
+        if request.namespace.is_empty() || id.is_empty() {
+            return Err(StreamError::invalid_argument(
+                "a delete needs the namespace and the owner's id of its kind",
+            ));
         }
-        if request.namespace.is_empty() || request.workflow_id.is_empty() {
+        if !other.is_empty() {
             return Err(StreamError::invalid_argument(
-                "a delete needs the namespace and the owner's Workflow id",
+                "a delete names the owner's id of its kind only",
             ));
         }
         self.store.delete_owner(request).await
     }
 
     /// The chain a producer writes on. A producer attempt stays on the chain it began on, and
-    /// is refused once a new chain took the Workflow id.
+    /// is refused once a new chain took the owner id.
     async fn producer_chain(&self, key: ProducerKey, writer: Writer<'_>) -> StreamResult<ChainId> {
         let chain = self.owner_chain(&key.owner, writer).await?;
         if let Some(began) = self.producers.get(&key)
@@ -763,11 +787,11 @@ impl Streams {
             // The latest run names the chain and tells whether it ended, in one call.
             let latest = self.describe(owner, "").await;
             if let (Err(OwnerError::NotFound(_)), Some(checked)) = (&latest, &known) {
-                // The chain came from this Workflow, so History has since dropped it.
+                // The chain came from this owner, so Temporal has since dropped it.
                 let gone = chain_of_key(owner, &checked.first_run_id);
                 return Err(self.gone(&gone, writer).await);
             }
-            let latest = latest.map_err(not_found)?;
+            let latest = latest.map_err(|error| not_found(owner, error))?;
             let chain = chain_of_key(owner, &first_run_id(owner, &latest)?);
             if latest.chain_ended() {
                 return Err(self.closed(&chain, "has closed", writer).await);
@@ -782,10 +806,10 @@ impl Streams {
             let pinned = self
                 .describe(owner, &owner.run_id)
                 .await
-                .map_err(not_found)?;
+                .map_err(|error| not_found(owner, error))?;
             let chain = chain_of_key(owner, &first_run_id(owner, &pinned)?);
             // A running run is its chain's latest, so it answers the end check too.
-            if pinned.status != WorkflowExecutionStatus::Running {
+            if pinned.status != OwnerStatus::Running {
                 self.refuse_if_ended(&chain, writer).await?;
             }
             chain
@@ -795,15 +819,11 @@ impl Streams {
     }
 
     /// Closes the chain and refuses when its latest run ended, when Temporal no longer holds it,
-    /// or when a new chain took its Workflow id.
+    /// or when a new chain took its owner id.
     async fn refuse_if_ended(&self, chain: &ChainId, writer: Writer<'_>) -> StreamResult<()> {
-        let key = OwnerKey {
-            namespace: chain.namespace.clone(),
-            workflow_id: chain.workflow_id.clone(),
-            run_id: String::new(),
-        };
+        let key = chain_owner_key(chain);
         match self.describe(&key, "").await {
-            // The chain came from this Workflow, so History has since dropped it.
+            // The chain came from this owner, so Temporal has since dropped it.
             Err(OwnerError::NotFound(_)) => Err(self.gone(chain, writer).await),
             Err(error) => Err(describe_failed(&error)),
             Ok(latest) if latest.chain_ended() || latest.first_run_id != chain.first_run_id => {
@@ -818,7 +838,8 @@ impl Streams {
     /// as a storage failure, which the caller retries.
     async fn gone(&self, chain: &ChainId, writer: Writer<'_>) -> StreamError {
         if self.gone_for_good(chain) {
-            self.closed(chain, "is gone from History", writer).await
+            self.closed(chain, gone_text(&chain_owner_key(chain)), writer)
+                .await
         } else {
             missing(chain)
         }
@@ -827,11 +848,7 @@ impl Streams {
     /// Whether Temporal answered that `chain`'s owner is gone twice, at least
     /// [StreamsOptions::owner_recheck] apart, with no answer that holds it in between.
     fn gone_for_good(&self, chain: &ChainId) -> bool {
-        let key = OwnerKey {
-            namespace: chain.namespace.clone(),
-            workflow_id: chain.workflow_id.clone(),
-            run_id: String::new(),
-        };
+        let key = chain_owner_key(chain);
         let recheck = self.options.owner_recheck;
         match self.missing.get(&key) {
             Some(seen) if seen.first_run_id == chain.first_run_id => seen.at.elapsed() >= recheck,
@@ -855,8 +872,8 @@ impl Streams {
             return error;
         }
         StreamError::closed(format!(
-            "the Workflow {:?} that owns this stream {how}",
-            chain.workflow_id
+            "the {} that owns this stream {how}",
+            chain_owner_key(chain).label()
         ))
     }
 
@@ -868,7 +885,10 @@ impl Streams {
         if !key.run_id.is_empty() {
             return self.chain(stream).await;
         }
-        let described = self.describe(&key, "").await.map_err(not_found)?;
+        let described = self
+            .describe(&key, "")
+            .await
+            .map_err(|error| not_found(&key, error))?;
         let first_run_id = first_run_id(&key, &described)?;
         self.chains
             .put(key, &first_run_id, self.options.owner_recheck);
@@ -879,13 +899,16 @@ impl Streams {
     pub(crate) async fn chain(&self, stream: &StreamAddress) -> StreamResult<ChainId> {
         let key = owner_key(stream);
         // A pinned run's chain never changes. The latest run's chain changes when a new chain
-        // takes the Workflow id.
+        // takes the owner id.
         if let Some(checked) = self.chains.get(&key)
             && (!key.run_id.is_empty() || checked.at.elapsed() < self.options.owner_recheck)
         {
             return Ok(chain_of(stream, &checked.first_run_id));
         }
-        let described = self.describe(&key, &key.run_id).await.map_err(not_found)?;
+        let described = self
+            .describe(&key, &key.run_id)
+            .await
+            .map_err(|error| not_found(&key, error))?;
         let first_run_id = first_run_id(&key, &described)?;
         self.chains
             .put(key, &first_run_id, self.options.owner_recheck);
@@ -899,25 +922,21 @@ impl Streams {
     ) -> Result<OwnerDescription, OwnerError> {
         let described = self
             .owner
-            .describe(&owner.namespace, &owner.workflow_id, run_id)
+            .describe(&owner.namespace, owner.owner(), run_id)
             .await;
         if described.is_ok() {
             self.missing.forget(&OwnerKey {
-                namespace: owner.namespace.clone(),
-                workflow_id: owner.workflow_id.clone(),
                 run_id: String::new(),
+                ..owner.clone()
             });
         }
         described
     }
 
     fn hash(&self, stream: &StreamAddress) -> String {
-        stream_hash(
-            &stream.namespace,
-            WORKFLOW_OWNER_KIND,
-            &stream.workflow_id,
-            &stream.topic,
-        )
+        let chain = chain_of(stream, "");
+        let (kind, id) = chain_owner(&chain);
+        stream_hash(&stream.namespace, kind, id, &stream.topic)
     }
 
     fn cursor(&self, stream: &StreamAddress, position: &str) -> String {
@@ -951,21 +970,44 @@ impl Streams {
 pub(crate) fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> {
     let stream =
         stream.ok_or_else(|| StreamError::invalid_argument("a stream call needs its stream"))?;
-    if stream.owner_kind != StreamOwnerKind::Workflow as i32 {
-        return Err(StreamError::unsupported(format!(
-            "this release keeps streams of Workflows only, not owner kind {}",
-            stream.owner_kind
-        )));
-    }
-    if stream.namespace.is_empty() || stream.workflow_id.is_empty() {
-        return Err(StreamError::invalid_argument(
-            "a stream needs its namespace and its owner's Workflow id",
-        ));
+    match StreamOwnerKind::try_from(stream.owner_kind) {
+        Ok(StreamOwnerKind::Workflow) => {
+            if stream.namespace.is_empty() || stream.workflow_id.is_empty() {
+                return Err(StreamError::invalid_argument(
+                    "a stream needs its namespace and its owner's Workflow id",
+                ));
+            }
+            if !stream.activity_id.is_empty() {
+                return Err(StreamError::invalid_argument(
+                    "a stream that a Workflow owns names no Activity id",
+                ));
+            }
+        }
+        Ok(StreamOwnerKind::Activity) => {
+            if stream.namespace.is_empty() || stream.activity_id.is_empty() {
+                return Err(StreamError::invalid_argument(
+                    "a stream needs its namespace and its owner's Activity id",
+                ));
+            }
+            if !stream.workflow_id.is_empty() {
+                return Err(StreamError::invalid_argument(
+                    "a stream that an Activity owns names no Workflow id",
+                ));
+            }
+        }
+        _ => return Err(unsupported_kind(stream.owner_kind)),
     }
     check_topic(&stream.topic)?;
     Ok(stream)
 }
 
+fn unsupported_kind(owner_kind: i32) -> StreamError {
+    StreamError::unsupported(format!(
+        "this release keeps streams of Workflows and standalone Activities only, not owner kind \
+         {owner_kind}"
+    ))
+}
+
 /// Refuses a topic no store keys reliably, by the rules Core's Workflow path also applies.
 /// The producer id and attempt a named producer or an Activity attempt writes as.
 fn producer(
@@ -988,7 +1030,7 @@ fn producer(
             }
             (named.producer_id.clone(), named.attempt)
         }
-        (None, Some(activity)) => (activity_producer_id(activity), i64::from(activity.attempt)),
+        (None, Some(activity)) => (activity_producer_id(activity), activity_attempt(activity)?),
         (None, None) => return Err(StreamError::invalid_argument("a call needs its producer")),
     };
     if attempt < 1 {
@@ -1037,43 +1079,83 @@ fn millis(duration: Duration) -> u64 {
 }
 
 fn owner_key(stream: &StreamAddress) -> OwnerKey {
+    let activity = stream.owner_kind == StreamOwnerKind::Activity as i32;
     OwnerKey {
         namespace: stream.namespace.clone(),
-        workflow_id: stream.workflow_id.clone(),
+        activity,
+        id: if activity {
+            stream.activity_id.clone()
+        } else {
+            stream.workflow_id.clone()
+        },
         run_id: stream.run_id.clone(),
     }
 }
 
-fn chain_of(stream: &StreamAddress, first_run_id: &str) -> ChainId {
+/// The latest run of the owner of `chain`.
+fn chain_owner_key(chain: &ChainId) -> OwnerKey {
+    let activity = chain.owner_kind == StreamOwnerKind::Activity as i32;
+    OwnerKey {
+        namespace: chain.namespace.clone(),
+        activity,
+        id: chain_owner(chain).1.to_string(),
+        run_id: String::new(),
+    }
+}
+
+/// The chain of `stream`'s owner that starts at `first_run_id`.
+pub(crate) fn chain_of(stream: &StreamAddress, first_run_id: &str) -> ChainId {
     chain_of_key(&owner_key(stream), first_run_id)
 }
 
 fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId {
-    ChainId {
-        namespace: owner.namespace.clone(),
-        workflow_id: owner.workflow_id.clone(),
-        first_run_id: first_run_id.to_string(),
+    // A Workflow chain keeps the kind unset, as every chain from before the Activity kind does,
+    // so the two compare equal.
+    if owner.activity {
+        ChainId {
+            namespace: owner.namespace.clone(),
+            first_run_id: first_run_id.to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: owner.id.clone(),
+            ..Default::default()
+        }
+    } else {
+        ChainId {
+            namespace: owner.namespace.clone(),
+            workflow_id: owner.id.clone(),
+            first_run_id: first_run_id.to_string(),
+            ..Default::default()
+        }
+    }
+}
+
+/// How a refusal says that Temporal no longer holds the owner.
+fn gone_text(owner: &OwnerKey) -> &'static str {
+    if owner.activity {
+        "is gone from Temporal"
+    } else {
+        "is gone from History"
     }
 }
 
 fn first_run_id(owner: &OwnerKey, described: &OwnerDescription) -> StreamResult<String> {
     if described.first_run_id.is_empty() {
-        // The Workflow's own output is keyed by the chain's first run, so an empty one would
-        // split the stream in two without an error.
+        // The owner's own output is keyed by the chain's first run, so an empty one would split
+        // the stream in two without an error.
         return Err(StreamError::unsupported(format!(
-            "Temporal did not report the first run id of Workflow {:?}, which streams are \
-             keyed by",
-            owner.workflow_id
+            "Temporal did not report the first run id of {}, which streams are keyed by",
+            owner.label()
         )));
     }
     Ok(described.first_run_id.clone())
 }
 
-fn not_found(error: OwnerError) -> StreamError {
+fn not_found(owner: &OwnerKey, error: OwnerError) -> StreamError {
     match error {
         OwnerError::NotFound(message) => StreamError::not_found(format!(
-            "the Workflow that owns the stream was not found, and its stream is keyed by its \
-             run chain: {message}"
+            "the {} that owns the stream was not found, and its stream is keyed by its run \
+             chain: {message}",
+            owner.label()
         )),
         error => describe_failed(&error),
     }
@@ -1081,9 +1163,9 @@ fn not_found(error: OwnerError) -> StreamError {
 
 fn missing(chain: &ChainId) -> StreamError {
     StreamError::storage(format!(
-        "Temporal doesn't hold the Workflow {:?} that owns this stream for now, so the call \
-         fails until it holds it again or answers the same after a recheck",
-        chain.workflow_id
+        "Temporal doesn't hold the owner {:?} of this stream for now, so the call fails until \
+         it holds it again or answers the same after a recheck",
+        chain_owner_key(chain).id
     ))
 }
 
@@ -1107,10 +1189,12 @@ mod tests {
     use sha2::{Digest, Sha256};
     use std::sync::atomic::{AtomicUsize, Ordering};
     use temporalio_common::protos::temporal::api::{
-        common::v1::Payload, enums::v1::WorkflowExecutionStatus,
+        common::v1::Payload,
+        enums::v1::{ActivityExecutionStatus, WorkflowExecutionStatus},
     };
 
-    /// Temporal as the tests need it: the latest run of each Workflow id, and pinned runs.
+    /// Temporal as the tests need it: the latest run of each owner, and pinned runs. A Workflow
+    /// owner is keyed by its id, and an Activity owner by `activity/<id>`.
     #[derive(Default)]
     struct Owners {
         latest: Mutex<HashMap<String, Result<OwnerDescription, OwnerError>>>,
@@ -1123,7 +1207,7 @@ mod tests {
             let described = OwnerDescription {
                 run_id: run_id.to_string(),
                 first_run_id: first_run_id.to_string(),
-                status: WorkflowExecutionStatus::Running,
+                status: OwnerStatus::Running,
             };
             self.runs
                 .lock()
@@ -1135,8 +1219,28 @@ mod tests {
                 .insert(workflow_id.to_string(), Ok(described));
         }
 
+        /// Starts run `run_id` of Activity `activity_id`, which is its own chain.
+        fn activity_run(&self, activity_id: &str, run_id: &str) {
+            self.run(&format!("activity/{activity_id}"), run_id, run_id);
+        }
+
         fn end(&self, workflow_id: &str, status: WorkflowExecutionStatus) {
-            if let Some(Ok(latest)) = self.latest.lock().unwrap().get_mut(workflow_id) {
+            self.end_as(workflow_id, OwnerStatus::of_workflow(status));
+        }
+
+        fn end_activity(&self, activity_id: &str, status: ActivityExecutionStatus) {
+            let status = OwnerStatus::of_activity(status);
+            let key = format!("activity/{activity_id}");
+            if let Some(Ok(latest)) = self.latest.lock().unwrap().get(&key)
+                && let Some(run) = self.runs.lock().unwrap().get_mut(&latest.run_id)
+            {
+                run.status = status;
+            }
+            self.end_as(&key, status);
+        }
+
+        fn end_as(&self, key: &str, status: OwnerStatus) {
+            if let Some(Ok(latest)) = self.latest.lock().unwrap().get_mut(key) {
                 latest.status = status;
             }
         }
@@ -1158,10 +1262,14 @@ mod tests {
         async fn describe(
             &self,
             _namespace: &str,
-            workflow_id: &str,
+            owner: OwnerId<'_>,
             run_id: &str,
         ) -> Result<OwnerDescription, OwnerError> {
             self.describes.fetch_add(1, Ordering::Relaxed);
+            let workflow_id = &match owner {
+                OwnerId::Workflow(id) => id.to_string(),
+                OwnerId::Activity(id) => format!("activity/{id}"),
+            };
             if !run_id.is_empty() {
                 return self
                     .runs
@@ -1217,6 +1325,7 @@ mod tests {
             workflow_id: "wf".to_string(),
             run_id: String::new(),
             topic: topic.to_string(),
+            ..Default::default()
         }
     }
 
@@ -1249,6 +1358,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 
@@ -1312,6 +1422,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
@@ -1657,16 +1768,36 @@ mod tests {
             OwnerDescription {
                 run_id: String::new(),
                 first_run_id: String::new(),
-                status,
+                status: OwnerStatus::of_workflow(status),
             }
             .chain_ended()
         };
         for status in [Completed, Failed, Canceled, Terminated, TimedOut] {
             assert!(ended(status), "{status:?}");
         }
-        for status in [Running, ContinuedAsNew, Unspecified] {
+        for status in [Running, ContinuedAsNew, Unspecified, Paused] {
             assert!(!ended(status), "{status:?}");
         }
+        assert_eq!(OwnerStatus::of_workflow(Paused), OwnerStatus::Running);
+    }
+
+    #[test]
+    fn only_a_terminal_activity_status_ends_its_run() {
+        use ActivityExecutionStatus::*;
+        for status in [Completed, Failed, Canceled, Terminated, TimedOut] {
+            assert_eq!(
+                OwnerStatus::of_activity(status),
+                OwnerStatus::Ended,
+                "{status:?}"
+            );
+        }
+        for status in [Running, Paused, Unspecified] {
+            assert_eq!(
+                OwnerStatus::of_activity(status),
+                OwnerStatus::Running,
+                "{status:?}"
+            );
+        }
     }
 
     #[tokio::test]
@@ -2400,7 +2531,7 @@ mod tests {
             .unwrap()
             .get_mut("run-1")
             .unwrap()
-            .status = WorkflowExecutionStatus::ContinuedAsNew;
+            .status = OwnerStatus::ContinuedAsNew;
         setup.owners.run("wf", "run-2", "run-1");
         let mut request = read_after("", &[], Duration::from_secs(5));
         request.stream.as_mut().unwrap().run_id = "run-1".to_string();
@@ -2603,6 +2734,7 @@ mod tests {
             namespace: "ns".to_string(),
             owner_kind: StreamOwnerKind::Workflow as i32,
             workflow_id: "wf".to_string(),
+            ..Default::default()
         };
         assert!(
             setup
@@ -2691,4 +2823,501 @@ mod tests {
         }
         assert_eq!(setup.owners.describes(), 0);
     }
+
+    /// A stream of standalone Activity `act`, whose run `act-run-1` is running.
+    fn activity_stream(topic: &str) -> StreamAddress {
+        StreamAddress {
+            namespace: "ns".to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: "act".to_string(),
+            topic: topic.to_string(),
+            ..Default::default()
+        }
+    }
+
+    fn activity_chain(run_id: &str) -> ChainId {
+        ChainId {
+            namespace: "ns".to_string(),
+            first_run_id: run_id.to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: "act".to_string(),
+            ..Default::default()
+        }
+    }
+
+    fn activity_setup() -> Setup {
+        let setup = setup(quick());
+        setup.owners.activity_run("act", "act-run-1");
+        setup
+    }
+
+    fn activity_append(sequence: i64, value: &str) -> AppendRequest {
+        AppendRequest {
+            stream: Some(activity_stream("out")),
+            producer: Some(Producer::Activity(ActivityProducer {
+                run_id: "act-run-1".to_string(),
+                activity_id: "act".to_string(),
+                attempt: 1,
+                ..Default::default()
+            })),
+            ..append("", sequence, value)
+        }
+    }
+
+    fn activity_read(after: &str, state: &[u8], wait: Duration) -> ReadRequest {
+        ReadRequest {
+            stream: Some(activity_stream("out")),
+            ..read_after(after, state, wait)
+        }
+    }
+
+    async fn read_activity_to_end(
+        streams: &Streams,
+        request: ReadRequest,
+    ) -> (usize, ReadResponse) {
+        let mut delivered = 0;
+        let mut request = request;
+        for _ in 0..50 {
+            let page = streams.read(request.clone()).await.unwrap();
+            delivered += page.records.len();
+            if page.done {
+                return (delivered, page);
+            }
+            request.after = page.cursor;
+            request.state = page.state;
+        }
+        panic!("the read never ended");
+    }
+
+    async fn activity_chain_closed(store: &MemoryStore, run_id: &str) -> bool {
+        store
+            .read(StoreReadRequest {
+                chain: Some(activity_chain(run_id)),
+                topic: "out".to_string(),
+                max_records: 1,
+                ..Default::default()
+            })
+            .await
+            .unwrap()
+            .closed
+    }
+
+    #[tokio::test]
+    async fn an_activity_writes_its_own_stream_on_the_chain_of_its_run() {
+        let setup = activity_setup();
+        let landed = setup.streams.append(activity_append(1, "1")).await.unwrap();
+        let held = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(activity_chain("act-run-1")),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(held.position, "0");
+        let read = setup
+            .streams
+            .read(activity_read("", &[], Duration::ZERO))
+            .await
+            .unwrap();
+        assert_eq!(read.records.len(), 1);
+        assert_eq!(read.records[0].cursor, landed.last_cursor);
+        assert_eq!(read.chain_first_run_id, "act-run-1");
+        // A Workflow with the same id owns other streams, and their cursors don't mix.
+        setup.owners.run("act", "run-1", "run-1");
+        let mut workflow = stream("out");
+        workflow.workflow_id = "act".to_string();
+        let workflow_read = setup
+            .streams
+            .read(ReadRequest {
+                stream: Some(workflow.clone()),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert!(workflow_read.records.is_empty());
+        let error = setup
+            .streams
+            .read(ReadRequest {
+                stream: Some(workflow),
+                after: landed.last_cursor,
+                ..Default::default()
+            })
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Cursor, "{error}");
+    }
+
+    #[tokio::test]
+    async fn an_activity_address_names_the_activity_only() {
+        let setup = activity_setup();
+        let mut no_id = activity_stream("out");
+        no_id.activity_id.clear();
+        let mut with_workflow = activity_stream("out");
+        with_workflow.workflow_id = "wf".to_string();
+        let mut workflow_with_activity = stream("out");
+        workflow_with_activity.activity_id = "act".to_string();
+        for bad in [no_id, with_workflow, workflow_with_activity] {
+            assert_eq!(
+                kind(
+                    setup
+                        .streams
+                        .latest(LatestRequest { stream: Some(bad) })
+                        .await
+                ),
+                StreamFailureKind::InvalidArgument
+            );
+        }
+        assert_eq!(setup.owners.describes(), 0);
+    }
+
+    #[tokio::test]
+    async fn an_activity_read_ends_done_when_the_run_reaches_a_terminal_status() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        setup
+            .owners
+            .end_activity("act", ActivityExecutionStatus::Failed);
+        let (delivered, end) = read_activity_to_end(
+            &setup.streams,
+            activity_read("", &[], Duration::from_secs(5)),
+        )
+        .await;
+        assert_eq!(delivered, 1);
+        assert!(end.done);
+        // A reader writes nothing, so the producers' owner check closes the chain.
+        assert!(!activity_chain_closed(&setup.store, "act-run-1").await);
+    }
+
+    #[tokio::test]
+    async fn an_activity_read_that_retries_and_pauses_stays_open() {
+        let setup = activity_setup();
+        setup
+            .owners
+            .end_activity("act", ActivityExecutionStatus::Paused);
+        let idle = setup
+            .streams
+            .read(activity_read("", &[], Duration::from_millis(200)))
+            .await
+            .unwrap();
+        assert!(!idle.done);
+        assert!(!activity_chain_closed(&setup.store, "act-run-1").await);
+    }
+
+    #[tokio::test]
+    async fn an_activity_read_before_the_run_exists_waits_the_grace() {
+        let setup = before_owner_starts(Duration::from_secs(60));
+        let empty = setup
+            .streams
+            .read(activity_read(END, &[], Duration::ZERO))
+            .await
+            .unwrap();
+        assert!(empty.records.is_empty() && !empty.done);
+        setup.owners.activity_run("act", "act-run-1");
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        let read = setup
+            .streams
+            .read(activity_read(&empty.cursor, &empty.state, Duration::ZERO))
+            .await
+            .unwrap();
+        assert_eq!(read.records.len(), 1);
+        let late = before_owner_starts(Duration::ZERO);
+        let error = late
+            .streams
+            .read(activity_read("", &[], Duration::from_secs(5)))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::NotFound);
+        assert!(error.message.contains("Activity \"act\""), "{error}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn an_activity_producer_is_refused_once_the_run_ended() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        setup
+            .owners
+            .end_activity("act", ActivityExecutionStatus::Completed);
+        tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await;
+        let error = setup
+            .streams
+            .append(activity_append(2, "2"))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Closed);
+        assert!(error.message.contains("Activity \"act\""), "{error}");
+        assert!(activity_chain_closed(&setup.store, "act-run-1").await);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_new_run_of_the_activity_id_ends_the_earlier_runs_streams() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        setup.owners.activity_run("act", "act-run-2");
+        tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await;
+        assert_eq!(
+            kind(setup.streams.append(activity_append(2, "2")).await),
+            StreamFailureKind::Closed
+        );
+        assert!(activity_chain_closed(&setup.store, "act-run-1").await);
+        // A producer of the new run writes on the new run's chain.
+        let mut next = activity_append(1, "3");
+        if let Some(Producer::Activity(activity)) = &mut next.producer {
+            activity.run_id = "act-run-2".to_string();
+        }
+        setup.streams.append(next).await.unwrap();
+        assert!(!activity_chain_closed(&setup.store, "act-run-2").await);
+    }
+
+    #[tokio::test]
+    async fn an_activity_read_pinned_to_a_run_reads_that_run() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        setup.owners.activity_run("act", "act-run-2");
+        let mut pinned = activity_read("", &[], Duration::ZERO);
+        pinned.stream.as_mut().unwrap().run_id = "act-run-1".to_string();
+        let page = setup.streams.read(pinned).await.unwrap();
+        assert_eq!(page.records.len(), 1);
+        assert_eq!(page.chain_first_run_id, "act-run-1");
+    }
+
+    #[tokio::test]
+    async fn an_activity_read_follows_the_newest_run_at_its_start_until_a_newer_run() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        let first = setup
+            .streams
+            .read(activity_read("", &[], Duration::ZERO))
+            .await
+            .unwrap();
+        assert_eq!(first.chain_first_run_id, "act-run-1");
+        setup.owners.activity_run("act", "act-run-2");
+        let (delivered, end) = read_activity_to_end(
+            &setup.streams,
+            activity_read(&first.cursor, &first.state, Duration::from_secs(5)),
+        )
+        .await;
+        assert_eq!(delivered, 0);
+        assert!(end.done);
+        assert!(!activity_chain_closed(&setup.store, "act-run-1").await);
+    }
+
+    #[tokio::test]
+    async fn an_activity_owner_delete_removes_only_that_activitys_streams() {
+        let setup = activity_setup();
+        setup.streams.append(activity_append(1, "1")).await.unwrap();
+        setup.streams.append(append("p", 1, "1")).await.unwrap();
+        let request = DeleteOwnerRequest {
+            namespace: "ns".to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: "act".to_string(),
+            ..Default::default()
+        };
+        assert!(
+            setup
+                .streams
+                .delete_owner(request.clone())
+                .await
+                .unwrap()
+                .deleted
+                > 0
+        );
+        let gone = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(activity_chain("act-run-1")),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(gone.position, "");
+        let kept = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(chain("run-1")),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(kept.position, "0");
+        for bad in [
+            DeleteOwnerRequest {
+                activity_id: String::new(),
+                ..request.clone()
+            },
+            DeleteOwnerRequest {
+                workflow_id: "wf".to_string(),
+                ..request.clone()
+            },
+        ] {
+            assert_eq!(
+                kind(setup.streams.delete_owner(bad).await),
+                StreamFailureKind::InvalidArgument
+            );
+        }
+    }
+
+    /// The start of 2026-10-14 in Unix milliseconds.
+    const DAY_START_MS: i64 = 1_791_936_000_000;
+
+    /// Attempt `attempt` of Activity `fetch` of run `run-1` of `wf`, started `after_ms` past
+    /// [DAY_START_MS], writing `value` at `sequence` on the Workflow's stream.
+    fn timed(attempt: i32, after_ms: i64, sequence: i64, value: &str) -> AppendRequest {
+        let mut activity = ActivityProducer {
+            workflow_id: "wf".to_string(),
+            run_id: "run-1".to_string(),
+            activity_id: "fetch".to_string(),
+            attempt,
+            ..Default::default()
+        };
+        let started = activity
+            .attempt_started_time
+            .get_or_insert_with(Default::default);
+        let ms = DAY_START_MS + after_ms;
+        started.seconds = ms / 1000;
+        started.nanos = (ms % 1000) as i32 * 1_000_000;
+        AppendRequest {
+            producer: Some(Producer::Activity(activity)),
+            ..append("", sequence, value)
+        }
+    }
+
+    fn key(attempt: i64, after_ms: i64) -> i64 {
+        (DAY_START_MS + after_ms) << 11 | attempt
+    }
+
+    /// The read's records, as each stored record's value and attempt, or a supersession's two
+    /// attempts with an empty value.
+    async fn shape(
+        setup: &Setup,
+        after: &str,
+        state: &[u8],
+    ) -> (Vec<(String, i64, i64, bool)>, ReadResponse) {
+        let page = setup
+            .streams
+            .read(read_after(after, state, Duration::ZERO))
+            .await
+            .unwrap();
+        let shape = page
+            .records
+            .iter()
+            .map(|record| match &record.record {
+                Some(read_record::Record::Stored(stored)) => (
+                    String::from_utf8(stored.body.clone().unwrap().data).unwrap(),
+                    stored.attempt,
+                    0,
+                    record.stale,
+                ),
+                Some(read_record::Record::Superseded(superseded)) => (
+                    String::new(),
+                    superseded.previous_attempt,
+                    superseded.attempt,
+                    record.stale,
+                ),
+                None => panic!("an empty record"),
+            })
+            .collect();
+        (shape, page)
+    }
+
+    #[tokio::test]
+    async fn a_timed_activity_attempt_writes_its_start_and_attempt_as_its_key() {
+        let setup = setup(quick());
+        setup.streams.append(timed(3, 0, 1, "a")).await.unwrap();
+        let (records, _) = shape(&setup, "", &[]).await;
+        assert_eq!(records, [("a".to_string(), key(3, 0), 0, false)]);
+        let mut late = timed(70_000, 0, 1, "b");
+        if let Some(Producer::Activity(activity)) = &mut late.producer {
+            activity.activity_id = "other".to_string();
+        }
+        setup.streams.append(late).await.unwrap();
+        let (records, _) = shape(&setup, "", &[]).await;
+        assert_eq!(records[1].1, key(crate::MAX_KEYED_ACTIVITY_ATTEMPT, 0));
+        assert_eq!(crate::MAX_KEYED_ACTIVITY_ATTEMPT, 2047);
+        // A JSON number holds the key exactly, so the Nexus contract's models take it.
+        assert!(records[1].1 < 1 << 53);
+        let before_1970 = timed(1, -DAY_START_MS - 1000, 1, "c");
+        // 2^42 ms is in May 2109, and a key past it would need more than 53 bits.
+        let after_2109 = timed(1, (1 << 42) - DAY_START_MS, 1, "d");
+        for refused in [before_1970, after_2109] {
+            assert_eq!(
+                kind(setup.streams.append(refused).await),
+                StreamFailureKind::InvalidArgument
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn an_activity_retry_supersedes_the_earlier_attempt() {
+        let setup = setup(quick());
+        setup.streams.append(timed(1, 0, 1, "a1")).await.unwrap();
+        setup
+            .streams
+            .append(timed(2, 1_000, 1, "b1"))
+            .await
+            .unwrap();
+        let (records, _) = shape(&setup, "", &[]).await;
+        assert_eq!(
+            records,
+            [
+                ("a1".to_string(), key(1, 0), 0, false),
+                (String::new(), key(1, 0), key(2, 1_000), false),
+                ("b1".to_string(), key(2, 1_000), 0, false),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn a_reset_activity_attempt_supersedes_with_no_divergence_and_no_drop() {
+        let setup = setup(quick());
+        for (attempt, after_ms) in [(1, 0), (2, 1_000), (3, 2_000)] {
+            setup
+                .streams
+                .append(timed(attempt, after_ms, 1, &format!("attempt {attempt}")))
+                .await
+                .unwrap();
+        }
+        // A reset sets the Temporal attempt back to 1, and the reset attempt starts later.
+        setup
+            .streams
+            .append(timed(1, 3_000, 1, "reset"))
+            .await
+            .unwrap();
+        let (records, _) = shape(&setup, "", &[]).await;
+        assert_eq!(
+            records.last().unwrap(),
+            &("reset".to_string(), key(1, 3_000), 0, false)
+        );
+        assert_eq!(
+            records[records.len() - 2],
+            (String::new(), key(3, 2_000), key(1, 3_000), false)
+        );
+        // With no start time the reset attempt would repeat attempt 1 with other content.
+        let mut untimed = timed(1, 0, 1, "reset");
+        if let Some(Producer::Activity(activity)) = &mut untimed.producer {
+            activity.attempt_started_time = None;
+        }
+        setup.streams.append(untimed.clone()).await.unwrap();
+        untimed.digest = vec![1; 32];
+        assert_eq!(
+            kind(setup.streams.append(untimed).await),
+            StreamFailureKind::ProducerDivergent
+        );
+    }
+
+    #[tokio::test]
+    async fn an_activity_attempt_that_writes_after_a_newer_one_started_is_stale() {
+        let setup = setup(quick());
+        setup.streams.append(timed(1, 0, 1, "a1")).await.unwrap();
+        setup
+            .streams
+            .append(timed(2, 1_000, 1, "b1"))
+            .await
+            .unwrap();
+        let (_, page) = shape(&setup, "", &[]).await;
+        // The first attempt timed out but keeps writing.
+        setup.streams.append(timed(1, 0, 2, "a2")).await.unwrap();
+        let (records, _) = shape(&setup, &page.cursor, &page.state).await;
+        assert_eq!(records, [("a2".to_string(), key(1, 0), 0, true)]);
+    }
 }
diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs
index 1631834e67..69f6956e0a 100644
--- a/crates/streams/src/lib.rs
+++ b/crates/streams/src/lib.rs
@@ -2,11 +2,12 @@
 
 //! Stream stores for Temporal Workflows.
 //!
-//! A stream is a topic of a Workflow, kept in a store outside Temporal and keyed by the
-//! Workflow's run chain. This crate holds what every store and every SDK share: the
-//! [StreamStore] trait a store implements, the cursor format, the record helpers that stamp
-//! records, and the error every call fails with. Lang reaches a store through Core's
-//! `StreamService`, and Core's Worker appends a Workflow's own output through the same trait.
+//! A stream is a topic of a Workflow or a standalone Activity, kept in a store outside Temporal
+//! and keyed by the owner's run chain. An Activity run is a chain of one run. This crate holds
+//! what every store and every SDK share: the [StreamStore] trait a store implements, the cursor
+//! format, the record helpers that stamp records, and the error every call fails with. Lang
+//! reaches a store through Core's `StreamService`, and Core's Worker appends a Workflow's own
+//! output through the same trait.
 
 mod close;
 mod cursor;
@@ -39,14 +40,14 @@ pub use notify::{
     WorkflowAttachment, progress_counter,
 };
 pub use notifylog::{NotificationEntry, NotificationLog};
-pub use owner::{OwnerClient, OwnerDescription, OwnerError};
+pub use owner::{OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus};
 pub use reader::{
     DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page,
 };
 pub use record::{
-    WORKFLOW_OWNER_KIND, activity_producer_id, check_append_digest, check_record_limit,
-    check_store_limits, close_record, stored_append_record, topic_record, workflow_batch_digest,
-    workflow_producer_id,
+    ACTIVITY_OWNER_KIND, MAX_KEYED_ACTIVITY_ATTEMPT, WORKFLOW_OWNER_KIND, activity_attempt,
+    activity_producer_id, chain_owner, check_append_digest, check_record_limit, check_store_limits,
+    close_record, stored_append_record, topic_record, workflow_batch_digest, workflow_producer_id,
 };
 pub use service::StreamService;
 pub use store::{
diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs
index 92d9fa00c0..4ecc266e78 100644
--- a/crates/streams/src/memory.rs
+++ b/crates/streams/src/memory.rs
@@ -8,8 +8,9 @@
 //! [MemoryStore::with_capabilities] masks capabilities off, so Core's fallbacks run in tests.
 
 use crate::{
-    Capabilities, ConditionForms, MAX_BATCH_BYTES, MAX_RECORD_BYTES, NotificationEntry,
-    NotificationLog, StreamError, StreamResult, StreamStore,
+    ACTIVITY_OWNER_KIND, Capabilities, ConditionForms, MAX_BATCH_BYTES, MAX_RECORD_BYTES,
+    NotificationEntry, NotificationLog, StreamError, StreamResult, StreamStore,
+    WORKFLOW_OWNER_KIND, chain_owner,
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse,
         StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose,
@@ -45,12 +46,15 @@ const FULL: Capabilities = Capabilities {
     max_batch_bytes: MAX_BATCH_BYTES,
 };
 
-type ChainKey = (String, String, String);
+/// A chain's namespace, owner kind, owner id and first run.
+type ChainKey = (String, &'static str, String, String);
 
 fn chain_key(chain: &ChainId) -> ChainKey {
+    let (kind, id) = chain_owner(chain);
     (
         chain.namespace.clone(),
-        chain.workflow_id.clone(),
+        kind,
+        id.to_string(),
         chain.first_run_id.clone(),
     )
 }
@@ -692,16 +696,21 @@ impl StreamStore for MemoryStore {
                 "this store can't delete the streams of an owner",
             ));
         }
-        if request.owner_kind != StreamOwnerKind::Workflow as i32 {
-            return Err(StreamError::unsupported(format!(
-                "this release keeps streams of Workflows only, not owner kind {}",
-                request.owner_kind
-            )));
-        }
+        let (owner_kind, owner_id) = match StreamOwnerKind::try_from(request.owner_kind) {
+            Ok(StreamOwnerKind::Workflow) => (WORKFLOW_OWNER_KIND, &request.workflow_id),
+            Ok(StreamOwnerKind::Activity) => (ACTIVITY_OWNER_KIND, &request.activity_id),
+            _ => {
+                return Err(StreamError::unsupported(format!(
+                    "this release keeps streams of Workflows and standalone Activities only, not \
+                     owner kind {}",
+                    request.owner_kind
+                )));
+            }
+        };
         let mut chains = self.chains.lock().unwrap();
         let mut deleted = 0;
-        chains.retain(|(namespace, workflow_id, _), chain| {
-            let owned = *namespace == request.namespace && *workflow_id == request.workflow_id;
+        chains.retain(|(namespace, kind, id, _), chain| {
+            let owned = *namespace == request.namespace && *kind == owner_kind && id == owner_id;
             if owned {
                 deleted += 1 + chain.topics.len() as u64;
             }
@@ -730,6 +739,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 
@@ -873,6 +883,7 @@ mod tests {
                 namespace: "ns".to_string(),
                 owner_kind: StreamOwnerKind::Unspecified as i32,
                 workflow_id: "wf".to_string(),
+                ..Default::default()
             })
             .await
             .unwrap_err();
diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs
index f2381ab47e..5b0cfd0ea7 100644
--- a/crates/streams/src/notify.rs
+++ b/crates/streams/src/notify.rs
@@ -4,7 +4,7 @@
 //! hint that there is something new to read. The records stay on the read path.
 
 use crate::{
-    CloseOutcome, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, mint_cursor,
+    CloseOutcome, StreamResult, StreamStore, chain_owner, mint_cursor,
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse,
         StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose,
@@ -214,6 +214,8 @@ struct StreamKey {
     namespace: String,
     workflow_id: String,
     first_run_id: String,
+    owner_kind: i32,
+    activity_id: String,
     topic: String,
 }
 
@@ -223,6 +225,8 @@ impl StreamKey {
             namespace: chain.namespace.clone(),
             workflow_id: chain.workflow_id.clone(),
             first_run_id: chain.first_run_id.clone(),
+            owner_kind: chain.owner_kind,
+            activity_id: chain.activity_id.clone(),
             topic: topic.to_string(),
         }
     }
@@ -232,6 +236,8 @@ impl StreamKey {
             namespace: self.namespace.clone(),
             workflow_id: self.workflow_id.clone(),
             first_run_id: self.first_run_id.clone(),
+            owner_kind: self.owner_kind,
+            activity_id: self.activity_id.clone(),
         }
     }
 }
@@ -342,6 +348,7 @@ impl Slot {
                 Err(NotifyError::Failed(error)) if tries_left > 0 && !self.superseded() => {
                     tracing::debug!(
                         workflow_id = %self.key.workflow_id,
+                        activity_id = %self.key.activity_id,
                         topic = %self.key.topic,
                         %error,
                         "A stream notification failed; trying again in {delay:?}"
@@ -350,6 +357,7 @@ impl Slot {
                 Err(error) => {
                     tracing::warn!(
                         workflow_id = %self.key.workflow_id,
+                        activity_id = %self.key.activity_id,
                         topic = %self.key.topic,
                         %error,
                         "A stream notification failed. The next write tells the reader again"
@@ -451,6 +459,7 @@ impl Notifier {
                     Err(NotifyError::Refused(error) | NotifyError::Unsupported(error)) => {
                         tracing::warn!(
                             workflow_id = %chain.workflow_id,
+                            activity_id = %chain.activity_id,
                             topic,
                             error,
                             "The server refused the close of a stream's notifier"
@@ -471,6 +480,7 @@ impl Notifier {
                     }
                     Err(NotifyError::Failed(error)) => tracing::warn!(
                         workflow_id = %chain.workflow_id,
+                        activity_id = %chain.activity_id,
                         topic,
                         error,
                         "Could not close a stream's notifier; trying again in {delay:?}"
@@ -684,13 +694,12 @@ impl NotifyingStore {
 }
 
 fn cursor_of(store: &dyn StreamStore, chain: &ChainId, topic: &str, position: &str) -> String {
-    let hash = stream_hash(
-        &chain.namespace,
-        WORKFLOW_OWNER_KIND,
-        &chain.workflow_id,
-        topic,
-    );
-    mint_cursor(store.name(), &hash, position)
+    let (kind, id) = chain_owner(chain);
+    mint_cursor(
+        store.name(),
+        &stream_hash(&chain.namespace, kind, id, topic),
+        position,
+    )
 }
 
 /// The cursor and counter that tell a reader about the record at `position`.
@@ -705,6 +714,7 @@ fn notification_at(
     let Some(counter) = progress_counter(position) else {
         tracing::warn!(
             workflow_id = %chain.workflow_id,
+            activity_id = %chain.activity_id,
             topic,
             position,
             "No notifier counter comes from this position, so the write isn't notified"
@@ -942,6 +952,18 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "wf".to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
+        }
+    }
+
+    /// The chain of run `run_id` of standalone Activity `wf`, which shares the Workflow's id.
+    fn activity_chain(run_id: &str) -> ChainId {
+        ChainId {
+            namespace: "ns".to_string(),
+            first_run_id: run_id.to_string(),
+            owner_kind: crate::proto::StreamOwnerKind::Activity as i32,
+            activity_id: "wf".to_string(),
+            ..Default::default()
         }
     }
 
@@ -1198,13 +1220,44 @@ mod tests {
             sent[0].position,
             mint_cursor(
                 "memory",
-                &stream_hash("ns", WORKFLOW_OWNER_KIND, "wf", "t"),
+                &stream_hash("ns", crate::WORKFLOW_OWNER_KIND, "wf", "t"),
                 &landed.last_position
             ),
             "the position is the cursor a reader resumes from"
         );
     }
 
+    #[tokio::test]
+    async fn an_activity_append_notifies_the_activitys_stream() {
+        let recorder = Arc::new(Recorder::default());
+        let (_, store) = notifying_store(&recorder);
+        // A Workflow with the same id writes too, and its stream is another one.
+        store
+            .append(append_request(&chain("run-1"), "t", 1))
+            .await
+            .unwrap();
+        let landed = store
+            .append(append_request(&activity_chain("run-1"), "t", 1))
+            .await
+            .unwrap();
+        store.flush_notifications().await;
+        let sent = recorder.sent.lock().unwrap().clone();
+        assert_eq!(sent.len(), 2, "one notifier per owner, {sent:?}");
+        let activity = sent
+            .iter()
+            .find(|n| n.chain == activity_chain("run-1"))
+            .unwrap();
+        assert_eq!(
+            activity.position,
+            mint_cursor(
+                "memory",
+                &stream_hash("ns", crate::ACTIVITY_OWNER_KIND, "wf", "t"),
+                &landed.last_position
+            ),
+            "the cursor hashes under the Activity kind"
+        );
+    }
+
     /// A store whose appends land at a position no counter comes from.
     struct OddPositions(MemoryStore);
 
@@ -1403,6 +1456,22 @@ mod tests {
         assert!(store.notifier().slots.lock().unwrap().is_empty());
     }
 
+    #[tokio::test]
+    async fn an_activity_close_reaches_its_notifier_with_the_outcome() {
+        let (client, store) = sees_the_store(None);
+        let run = activity_chain("act-run");
+        store.append(append_request(&run, "t", 1)).await.unwrap();
+        store
+            .close_stream(&run, "t", CloseOutcome::Failure(gave_up()).into())
+            .await
+            .unwrap();
+        let sent = client.sent.lock().unwrap().clone();
+        let close = sent.last().unwrap();
+        assert_eq!(close.chain, run);
+        assert_eq!(close.close, Some(CloseOutcome::Failure(gave_up())));
+        assert_eq!(close.counter, 2, "one above the newest record's counter");
+    }
+
     #[tokio::test]
     async fn a_second_close_tells_the_notifier_the_first() {
         // A rerun that closes again must not give waiters another outcome than readers get.
diff --git a/crates/streams/src/owner.rs b/crates/streams/src/owner.rs
index 37bd22ec4d..14ff04eaf5 100644
--- a/crates/streams/src/owner.rs
+++ b/crates/streams/src/owner.rs
@@ -1,41 +1,79 @@
 //! What the stream layer asks Temporal about a stream's owner.
 //!
-//! A store can't see a Workflow close, so the layer above it asks: to find the chain a stream is
+//! A store can't see an owner close, so the layer above it asks: to find the chain a stream is
 //! keyed by, to refuse a producer once the chain ended, and to end a read. Core implements this
 //! over its client, and tests use a fake.
 
-use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus;
+use temporalio_common::protos::temporal::api::enums::v1::{
+    ActivityExecutionStatus, WorkflowExecutionStatus,
+};
 
-/// One run of a Workflow, as Temporal describes it.
+/// The owner of a stream, by kind and id.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OwnerId<'a> {
+    /// A Workflow id.
+    Workflow(&'a str),
+    /// A standalone Activity id.
+    Activity(&'a str),
+}
+
+/// How an owner's run stands, as the stream layer needs it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OwnerStatus {
+    /// The run has not reached a terminal status.
+    Running,
+    /// The run closed, and its successor in the chain writes on.
+    ContinuedAsNew,
+    /// The run closed and ended its chain.
+    Ended,
+}
+
+impl OwnerStatus {
+    /// The status of a Workflow run. A paused run is running, since it can resume and write.
+    pub fn of_workflow(status: WorkflowExecutionStatus) -> Self {
+        use WorkflowExecutionStatus::*;
+        match status {
+            Completed | Failed | Canceled | Terminated | TimedOut => Self::Ended,
+            ContinuedAsNew => Self::ContinuedAsNew,
+            Unspecified | Running | Paused => Self::Running,
+        }
+    }
+
+    /// The status of a standalone Activity run. Retries, backoff, pause and reset keep the run, so
+    /// only a terminal status ends it.
+    pub fn of_activity(status: ActivityExecutionStatus) -> Self {
+        use ActivityExecutionStatus::*;
+        match status {
+            Completed | Failed | Canceled | Terminated | TimedOut => Self::Ended,
+            Unspecified | Running | Paused => Self::Running,
+        }
+    }
+}
+
+/// One run of an owner, as Temporal describes it.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct OwnerDescription {
     /// The run described.
     pub run_id: String,
-    /// The first run of the chain the run belongs to. Streams are keyed by it.
+    /// The first run of the chain the run belongs to. Streams are keyed by it. An Activity run is
+    /// its own chain.
     pub first_run_id: String,
     /// How the run stands.
-    pub status: WorkflowExecutionStatus,
+    pub status: OwnerStatus,
 }
 
 impl OwnerDescription {
     /// Whether the run ended its chain. A run that continued as new did not, since its successor
     /// writes on.
     pub fn chain_ended(&self) -> bool {
-        matches!(
-            self.status,
-            WorkflowExecutionStatus::Completed
-                | WorkflowExecutionStatus::Failed
-                | WorkflowExecutionStatus::Canceled
-                | WorkflowExecutionStatus::Terminated
-                | WorkflowExecutionStatus::TimedOut
-        )
+        self.status == OwnerStatus::Ended
     }
 }
 
 /// Why a describe failed.
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
 pub enum OwnerError {
-    /// Temporal holds no such Workflow or run, or no longer does.
+    /// Temporal holds no such owner or run, or no longer does.
     #[error("not found: {0}")]
     NotFound(String),
     /// Temporal could not answer.
@@ -46,12 +84,11 @@ pub enum OwnerError {
 /// Temporal, as the stream layer asks it about owners.
 #[async_trait::async_trait]
 pub trait OwnerClient: Send + Sync {
-    /// Describes run `run_id` of `workflow_id`, or the Workflow id's latest run when `run_id` is
-    /// empty.
+    /// Describes run `run_id` of `owner`, or the owner's latest run when `run_id` is empty.
     async fn describe(
         &self,
         namespace: &str,
-        workflow_id: &str,
+        owner: OwnerId<'_>,
         run_id: &str,
     ) -> Result<OwnerDescription, OwnerError>;
 }
diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs
index e431885c0a..346a022d28 100644
--- a/crates/streams/src/reader.rs
+++ b/crates/streams/src/reader.rs
@@ -766,6 +766,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"),
diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs
index 60b988f097..16a13be33c 100644
--- a/crates/streams/src/record.rs
+++ b/crates/streams/src/record.rs
@@ -2,7 +2,9 @@
 
 use crate::{
     Capabilities, StreamError, StreamResult,
-    proto::{ActivityProducer, AppendRecord, StreamRecord, StreamRecordKind},
+    proto::{
+        ActivityProducer, AppendRecord, ChainId, StreamOwnerKind, StreamRecord, StreamRecordKind,
+    },
 };
 use temporalio_common::{
     protos::temporal::api::common::v1::Payload,
@@ -12,6 +14,18 @@ use temporalio_common::{
 /// The owner kind a Workflow's streams hash under in their cursors.
 pub const WORKFLOW_OWNER_KIND: &str = "workflow";
 
+/// The owner kind a standalone Activity's streams hash under in their cursors.
+pub const ACTIVITY_OWNER_KIND: &str = "activity";
+
+/// The owner kind a chain's streams hash under, and the owner's id.
+pub fn chain_owner(chain: &ChainId) -> (&'static str, &str) {
+    if chain.owner_kind == StreamOwnerKind::Activity as i32 {
+        (ACTIVITY_OWNER_KIND, &chain.activity_id)
+    } else {
+        (WORKFLOW_OWNER_KIND, &chain.workflow_id)
+    }
+}
+
 const HASH_LENGTH: usize = 32;
 
 /// The producer id an Activity attempt writes as.
@@ -22,6 +36,49 @@ pub fn activity_producer_id(activity: &ActivityProducer) -> String {
     format!("{}@{}", activity.activity_id, activity.run_id)
 }
 
+/// The low bits of an Activity attempt key that hold the Temporal attempt.
+const ACTIVITY_ATTEMPT_BITS: u32 = 11;
+
+/// The high bits of an Activity attempt key that hold the start time in Unix milliseconds. The
+/// whole key fits 53 bits, so a JSON number holds it exactly.
+const ACTIVITY_START_BITS: u32 = 53 - ACTIVITY_ATTEMPT_BITS;
+
+/// The largest Temporal attempt an Activity attempt key holds. A higher attempt keeps this one.
+pub const MAX_KEYED_ACTIVITY_ATTEMPT: i64 = (1 << ACTIVITY_ATTEMPT_BITS) - 1;
+
+/// The attempt an Activity attempt writes as.
+///
+/// With `attempt_started_time`, it is `started_unix_ms << 11 | min(attempt, 2047)`. A retry and
+/// a reset attempt start later, so they write a higher attempt and supersede the earlier one, even
+/// though a reset sets the Temporal attempt back to 1. An attempt that writes on after a newer one
+/// started reads as stale. With no start time, it is the Temporal attempt, as a lang that sends
+/// none expects. Fails as an invalid argument for an attempt below one or a start time outside
+/// what the key holds.
+pub fn activity_attempt(activity: &ActivityProducer) -> StreamResult<i64> {
+    let attempt = i64::from(activity.attempt);
+    if attempt < 1 {
+        return Err(StreamError::invalid_argument(format!(
+            "a producer's attempt starts at one, not {attempt}"
+        )));
+    }
+    let Some(started) = &activity.attempt_started_time else {
+        return Ok(attempt);
+    };
+    let unix_ms = started
+        .seconds
+        .checked_mul(1000)
+        .and_then(|ms| ms.checked_add(i64::from(started.nanos.div_euclid(1_000_000))))
+        .filter(|ms| (1..1 << ACTIVITY_START_BITS).contains(ms))
+        .ok_or_else(|| {
+            StreamError::invalid_argument(format!(
+                "an Activity attempt's start time must be after 1970 and before the year 2109, \
+                 not {}s {}ns",
+                started.seconds, started.nanos
+            ))
+        })?;
+    Ok(unix_ms << ACTIVITY_ATTEMPT_BITS | attempt.min(MAX_KEYED_ACTIVITY_ATTEMPT))
+}
+
 /// The producer id a Workflow run writes its own records as.
 ///
 /// It is an Activity's id form with no Activity id. An Activity id is never empty, so no Activity
@@ -239,6 +296,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");
     }
diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs
index 81dc91a109..41f50a8fcb 100644
--- a/crates/streams/src/redis/keys.rs
+++ b/crates/streams/src/redis/keys.rs
@@ -11,6 +11,10 @@
 //! <prefix>:{<namespace>:<workflow id>:<first run id>}:chain            the close flag
 //! ```
 //!
+//! A standalone Activity run is a chain of one run. Its tag has the literal `!a` after the
+//! namespace, `{<namespace>:!a:<activity id>:<run id>}`, and the same keys follow it. Each part
+//! escapes `!`, so no Workflow id makes that tag.
+//!
 //! The notification log lives apart from the chains, one stream per Cluster slot and a few
 //! doorbells, each under the hash tag of a slot (see `slots.rs`). A chain's appends go to the
 //! stream of the chain's own slot, so the append script writes both:
@@ -25,7 +29,11 @@
 //! two streams share a key.
 
 use super::{conn::slot, slots};
-use crate::proto::ChainId;
+use crate::proto::{ChainId, StreamOwnerKind};
+
+/// The part after the namespace that marks an Activity chain's tag. [part] escapes `!`, so no
+/// Workflow id makes it.
+const ACTIVITY_MARK: &str = "!a";
 
 /// Percent-encodes every byte but the unreserved ones, as Python's `quote(text, safe="")` does.
 pub(crate) fn part(text: &str) -> String {
@@ -48,12 +56,16 @@ pub(crate) struct ChainKeys {
 
 impl ChainKeys {
     pub(crate) fn new(prefix: &str, chain: &ChainId) -> Self {
+        let owner = if chain.owner_kind == StreamOwnerKind::Activity as i32 {
+            format!("{ACTIVITY_MARK}:{}", part(&chain.activity_id))
+        } else {
+            part(&chain.workflow_id)
+        };
         Self {
             base: format!(
-                "{}:{{{}:{}:{}}}",
+                "{}:{{{}:{owner}:{}}}",
                 part(prefix),
                 part(&chain.namespace),
-                part(&chain.workflow_id),
                 part(&chain.first_run_id)
             ),
         }
@@ -114,19 +126,24 @@ pub(crate) fn doorbell(prefix: &str, doorbell: u32, doorbells: u32) -> String {
 pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)> {
     let rest = key.strip_prefix(&format!("{}:{{", part(prefix)))?;
     let (tag, topic) = rest.split_once("}:t:")?;
-    let mut parts = tag.split(':');
-    let (namespace, workflow_id, first_run_id) = (parts.next()?, parts.next()?, parts.next()?);
-    if parts.next().is_some() {
-        return None;
-    }
-    Some((
-        ChainId {
+    let parts: Vec<&str> = tag.split(':').collect();
+    let chain = match parts[..] {
+        [namespace, workflow_id, first_run_id] if workflow_id != ACTIVITY_MARK => ChainId {
             namespace: unpart(namespace)?,
             workflow_id: unpart(workflow_id)?,
             first_run_id: unpart(first_run_id)?,
+            ..Default::default()
         },
-        unpart(topic)?,
-    ))
+        [namespace, ACTIVITY_MARK, activity_id, run_id] => ChainId {
+            namespace: unpart(namespace)?,
+            first_run_id: unpart(run_id)?,
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: unpart(activity_id)?,
+            ..Default::default()
+        },
+        _ => return None,
+    };
+    Some((chain, unpart(topic)?))
 }
 
 /// Undoes [part].
@@ -162,6 +179,16 @@ pub(crate) fn owner_pattern(prefix: &str, namespace: &str, workflow_id: &str) ->
     )
 }
 
+/// The pattern that matches every key of one standalone Activity id, across all its runs.
+pub(crate) fn activity_owner_pattern(prefix: &str, namespace: &str, activity_id: &str) -> String {
+    format!(
+        "{}:{{{}:{ACTIVITY_MARK}:{}:*",
+        part(prefix),
+        part(namespace),
+        part(activity_id)
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -173,6 +200,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "w:f".to_string(),
             first_run_id: "run".to_string(),
+            ..Default::default()
         };
         let keys = ChainKeys::new("p", &chain);
         for topic in ["out", "a:b", "é {x}%"] {
@@ -187,21 +215,34 @@ mod tests {
         case[field].as_str().unwrap()
     }
 
+    fn case_chain(case: &Value) -> ChainId {
+        if case["owner_kind"] == "activity" {
+            ChainId {
+                namespace: text(case, "namespace").to_string(),
+                first_run_id: text(case, "first_run_id").to_string(),
+                owner_kind: StreamOwnerKind::Activity as i32,
+                activity_id: text(case, "activity_id").to_string(),
+                ..Default::default()
+            }
+        } else {
+            ChainId {
+                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()
+            }
+        }
+    }
+
     #[test]
     fn keys_match_the_python_sdk() {
         // Streams written by an SDK before the store moved into Core must stay readable.
         let cases: Vec<Value> =
             serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap();
         assert!(cases.len() >= 4);
+        assert!(cases.iter().any(|case| case["owner_kind"] == "activity"));
         for case in &cases {
-            let keys = ChainKeys::new(
-                text(case, "prefix"),
-                &ChainId {
-                    namespace: text(case, "namespace").to_string(),
-                    workflow_id: text(case, "workflow_id").to_string(),
-                    first_run_id: text(case, "first_run_id").to_string(),
-                },
-            );
+            let keys = ChainKeys::new(text(case, "prefix"), &case_chain(case));
             let topic = text(case, "topic");
             assert_eq!(keys.log(topic), text(case, "log"));
             assert_eq!(keys.meta(topic), text(case, "meta"));
@@ -214,11 +255,7 @@ mod tests {
         let cases: Vec<Value> =
             serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap();
         for case in &cases {
-            let chain = ChainId {
-                namespace: text(case, "namespace").to_string(),
-                workflow_id: text(case, "workflow_id").to_string(),
-                first_run_id: text(case, "first_run_id").to_string(),
-            };
+            let chain = case_chain(case);
             let prefix = text(case, "prefix");
             let keys = ChainKeys::new(prefix, &chain);
             let topic = text(case, "topic");
@@ -232,6 +269,8 @@ mod tests {
         }
         assert_eq!(chain_of_log("p", "q:{ns:wf:run}:t:out"), None);
         assert_eq!(chain_of_log("p", "p:{ns:wf}:t:out"), None);
+        assert_eq!(chain_of_log("p", "p:{ns:!b:act:run}:t:out"), None);
+        assert_eq!(chain_of_log("p", "p:{ns:!a:act}:t:out"), None);
     }
 
     #[test]
@@ -252,6 +291,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:*");
@@ -267,6 +307,44 @@ mod tests {
         assert_eq!(owner_pattern("p", "ns", "a*b"), "p:{ns:a%2Ab:*");
     }
 
+    #[test]
+    fn an_activity_chain_keeps_apart_from_every_workflow() {
+        let activity = |id: &str| ChainId {
+            namespace: "ns".to_string(),
+            first_run_id: "run-1".to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: id.to_string(),
+            ..Default::default()
+        };
+        let workflow = |id: &str| ChainId {
+            namespace: "ns".to_string(),
+            workflow_id: id.to_string(),
+            first_run_id: "run-1".to_string(),
+            ..Default::default()
+        };
+        let matcher =
+            |pattern: String| move |key: &str| key.starts_with(pattern.trim_end_matches('*'));
+        for id in ["wf", "!a", "a", "!a:wf"] {
+            let keys = ChainKeys::new("p", &activity(id));
+            for workflow_id in [id, "!a", "a", "%21a"] {
+                assert!(
+                    !matcher(owner_pattern("p", "ns", workflow_id))(&keys.chain()),
+                    "{id}"
+                );
+                assert_ne!(keys, ChainKeys::new("p", &workflow(workflow_id)), "{id}");
+            }
+            let own = matcher(activity_owner_pattern("p", "ns", id));
+            assert!(own(&keys.chain()) && own(&keys.log("out")), "{id}");
+            assert!(!own(&ChainKeys::new("p", &workflow(id)).chain()), "{id}");
+            // The `:` after the id keeps `wf` from matching the runs of `wf2`.
+            assert!(
+                !own(&ChainKeys::new("p", &activity(&format!("{id}2"))).chain()),
+                "{id}"
+            );
+        }
+        assert_eq!(activity_owner_pattern("p", "ns", "a*b"), "p:{ns:!a:a%2Ab:*");
+    }
+
     #[test]
     fn the_doorbells_spread_over_the_slot_range() {
         let slots: Vec<u16> = (0..16)
diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs
index 6d2bf288aa..77077c22da 100644
--- a/crates/streams/src/redis/mod.rs
+++ b/crates/streams/src/redis/mod.rs
@@ -18,7 +18,7 @@ use crate::{
 };
 use conn::{BlockingPool, Shared};
 use errors::{Call, Mapped};
-use keys::{ChainKeys, owner_pattern, session_field};
+use keys::{ChainKeys, activity_owner_pattern, owner_pattern, session_field};
 use prost::Message;
 use redis::{Cmd, Script, Value, streams::StreamReadReply};
 use std::{
@@ -974,15 +974,23 @@ impl StreamStore for RedisStore {
     }
 
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
-        if request.owner_kind != StreamOwnerKind::Workflow as i32 {
-            return Err(StreamError::unsupported(format!(
-                "this release keeps streams of Workflows only, not owner kind {}",
-                request.owner_kind
-            )));
-        }
-        // A Workflow id's chains hash to different slots, so every primary is scanned. Some keys
+        let pattern = match StreamOwnerKind::try_from(request.owner_kind) {
+            Ok(StreamOwnerKind::Workflow) => {
+                owner_pattern(&self.prefix, &request.namespace, &request.workflow_id)
+            }
+            Ok(StreamOwnerKind::Activity) => {
+                activity_owner_pattern(&self.prefix, &request.namespace, &request.activity_id)
+            }
+            _ => {
+                return Err(StreamError::unsupported(format!(
+                    "this release keeps streams of Workflows and standalone Activities only, not \
+                     owner kind {}",
+                    request.owner_kind
+                )));
+            }
+        };
+        // An owner id's chains hash to different slots, so every primary is scanned. Some keys
         // may be gone when a later step fails, which is why a failure leaves the outcome unknown.
-        let pattern = owner_pattern(&self.prefix, &request.namespace, &request.workflow_id);
         let mut deleted = 0;
         for node in self.shared.primaries().await.mapped(Call::Write)? {
             let mut cursor = "0".to_string();
@@ -1090,6 +1098,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: workflow_id.to_string(),
             first_run_id: "run-1".to_string(),
+            ..Default::default()
         }
     }
 
diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs
index f918fe2481..5ac53cd3ab 100644
--- a/crates/streams/src/service.rs
+++ b/crates/streams/src/service.rs
@@ -335,16 +335,14 @@ fn duration(seconds: i64, nanos: i32, name: &str) -> StreamResult<Duration> {
 mod tests {
     use super::*;
     use crate::{
-        OwnerDescription, OwnerError,
+        OwnerDescription, OwnerError, OwnerId, OwnerStatus,
         proto::{
             AppendRecord, AppendResponse, LatestResponse, MemoryStoreConfig, NamedProducer,
             ReadResponse, StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind,
             append_request, read_record,
         },
     };
-    use temporalio_common::protos::temporal::api::{
-        common::v1::Payload, enums::v1::WorkflowExecutionStatus,
-    };
+    use temporalio_common::protos::temporal::api::common::v1::Payload;
 
     /// Every Workflow's latest run is `run-1`, the first of its chain, and still running.
     struct Running;
@@ -354,13 +352,13 @@ mod tests {
         async fn describe(
             &self,
             _: &str,
-            _: &str,
+            _: OwnerId<'_>,
             _: &str,
         ) -> Result<OwnerDescription, OwnerError> {
             Ok(OwnerDescription {
                 run_id: "run-1".to_string(),
                 first_run_id: "run-1".to_string(),
-                status: WorkflowExecutionStatus::Running,
+                status: OwnerStatus::Running,
             })
         }
     }
@@ -386,6 +384,7 @@ mod tests {
             workflow_id: "wf".to_string(),
             run_id: String::new(),
             topic: "out".to_string(),
+            ..Default::default()
         }
     }
 
@@ -589,6 +588,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())
@@ -599,6 +599,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())
@@ -997,7 +998,7 @@ mod tests {
         async fn describe(
             &self,
             _: &str,
-            _: &str,
+            _: OwnerId<'_>,
             _: &str,
         ) -> Result<OwnerDescription, OwnerError> {
             let ended = self.0.load(std::sync::atomic::Ordering::SeqCst);
@@ -1005,9 +1006,9 @@ mod tests {
                 run_id: "run-1".to_string(),
                 first_run_id: "run-1".to_string(),
                 status: if ended {
-                    WorkflowExecutionStatus::Completed
+                    OwnerStatus::Ended
                 } else {
-                    WorkflowExecutionStatus::Running
+                    OwnerStatus::Running
                 },
             })
         }
diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs
index 8326d9b001..cd3f50d350 100644
--- a/crates/streams/src/subscribe.rs
+++ b/crates/streams/src/subscribe.rs
@@ -9,7 +9,7 @@
 use crate::{
     NotifierClient, NotifyError, OWNER_RECHECK, StreamError, StreamResult, Streams, StreamsOptions,
     WorkflowAttachment,
-    handle::{Remembered, address},
+    handle::{Remembered, address, chain_of},
     proto::{
         ChainId, StreamAddress, SubscribeRequest, SubscribeResponse, UnsubscribeRequest,
         UnsubscribeResponse, WaitForNotificationRequest, WaitForNotificationResponse,
@@ -41,7 +41,9 @@ impl Default for Subscriptions {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct SubscriptionKey {
     namespace: String,
+    owner_kind: i32,
     owner_workflow_id: String,
+    owner_activity_id: String,
     owner_run_id: String,
     topic: String,
     request_id: String,
@@ -51,7 +53,9 @@ impl SubscriptionKey {
     fn new(stream: &StreamAddress, request_id: &str) -> Self {
         Self {
             namespace: stream.namespace.clone(),
+            owner_kind: stream.owner_kind,
             owner_workflow_id: stream.workflow_id.clone(),
+            owner_activity_id: stream.activity_id.clone(),
             owner_run_id: stream.run_id.clone(),
             topic: stream.topic.clone(),
             request_id: request_id.to_string(),
@@ -179,11 +183,7 @@ impl Subscriptions {
         let client = self.client()?;
         let key = SubscriptionKey::new(stream, &request_id);
         let chain = match self.attached.get(&key) {
-            Some(attached) => crate::proto::ChainId {
-                namespace: stream.namespace.clone(),
-                workflow_id: stream.workflow_id.clone(),
-                first_run_id: attached.first_run_id,
-            },
+            Some(attached) => chain_of(stream, &attached.first_run_id),
             None => streams.chain_now(stream).await?,
         };
         client
@@ -248,13 +248,12 @@ impl Subscriptions {
 mod tests {
     use super::*;
     use crate::{
-        MemoryStore, NotifierPoll, OwnerClient, OwnerDescription, OwnerError, StreamNotification,
-        StreamService, StreamsOptions,
+        MemoryStore, NotifierPoll, OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus,
+        StreamNotification, StreamService, StreamsOptions,
         proto::{ChainId, StreamFailureKind, StreamOwnerKind},
     };
     use prost::Message;
     use std::sync::Mutex;
-    use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus;
 
     /// The owner's chain is `first_run_id`, which a test moves to a new chain.
     struct Owner(Mutex<String>);
@@ -264,14 +263,14 @@ mod tests {
         async fn describe(
             &self,
             _: &str,
-            _: &str,
+            _: OwnerId<'_>,
             _: &str,
         ) -> Result<OwnerDescription, OwnerError> {
             let first_run_id = self.0.lock().unwrap().clone();
             Ok(OwnerDescription {
                 run_id: first_run_id.clone(),
                 first_run_id,
-                status: WorkflowExecutionStatus::Running,
+                status: OwnerStatus::Running,
             })
         }
     }
@@ -363,6 +362,7 @@ mod tests {
             workflow_id: "owner".to_string(),
             run_id: String::new(),
             topic: "tokens".to_string(),
+            ..Default::default()
         }
     }
 
@@ -391,6 +391,7 @@ mod tests {
             namespace: "ns".to_string(),
             workflow_id: "owner".to_string(),
             first_run_id: first_run_id.to_string(),
+            ..Default::default()
         }
     }
 
@@ -610,6 +611,53 @@ mod tests {
         assert_eq!(answer.chain_first_run_id, "chain-2");
     }
 
+    #[tokio::test]
+    async fn a_workflow_reader_subscribes_to_a_stream_that_an_activity_owns() {
+        let setup = setup();
+        let activity = StreamAddress {
+            owner_kind: StreamOwnerKind::Activity as i32,
+            workflow_id: String::new(),
+            activity_id: "owner".to_string(),
+            ..stream()
+        };
+        let activity_chain = ChainId {
+            namespace: "ns".to_string(),
+            first_run_id: "chain-1".to_string(),
+            owner_kind: StreamOwnerKind::Activity as i32,
+            activity_id: "owner".to_string(),
+            ..Default::default()
+        };
+        let request = SubscribeRequest {
+            stream: Some(activity.clone()),
+            ..subscribe_request("1")
+        };
+        let answer = subscribe(&setup.service, request).await.unwrap();
+        assert!(answer.attached);
+        assert_eq!(answer.chain_first_run_id, "chain-1");
+        // A Workflow with the same id owns another stream, so its reader attaches apart.
+        assert!(
+            subscribe(&setup.service, subscribe_request("1"))
+                .await
+                .unwrap()
+                .attached
+        );
+        let attached = setup.server.attached.lock().unwrap().clone();
+        assert_eq!(attached[0].chain, activity_chain);
+        assert_eq!(attached[1].chain, chain("chain-1"));
+        let unsubscribe = UnsubscribeRequest {
+            stream: Some(activity),
+            run_id: "reader-run".to_string(),
+            subscription_id: "1".to_string(),
+            ..Default::default()
+        };
+        setup
+            .service
+            .call("Unsubscribe", &unsubscribe.encode_to_vec())
+            .await
+            .unwrap();
+        assert_eq!(setup.server.detached.lock().unwrap()[0].0, activity_chain);
+    }
+
     async fn wait(
         setup: &Setup,
         after_counter: i64,
@@ -641,6 +689,7 @@ mod tests {
                     namespace: "ns".to_string(),
                     workflow_id: "owner".to_string(),
                     first_run_id,
+                    ..Default::default()
                 }),
                 topic: "tokens".to_string(),
                 producer_id: "p".to_string(),
diff --git a/crates/streams/testdata/redis/keys.json b/crates/streams/testdata/redis/keys.json
index e6deb56472..23f823bbd4 100644
--- a/crates/streams/testdata/redis/keys.json
+++ b/crates/streams/testdata/redis/keys.json
@@ -48,5 +48,49 @@
     "log": "pre%20fix:{n%25s:a~b.c_d-e:r}:t:x%20y",
     "meta": "pre%20fix:{n%25s:a~b.c_d-e:r}:t:x%20y:meta",
     "chain": "pre%20fix:{n%25s:a~b.c_d-e:r}:chain"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "default",
+    "owner_kind": "activity",
+    "activity_id": "gen-1",
+    "first_run_id": "run-1",
+    "topic": "tokens",
+    "log": "temporal-streams:{default:!a:gen-1:run-1}:t:tokens",
+    "meta": "temporal-streams:{default:!a:gen-1:run-1}:t:tokens:meta",
+    "chain": "temporal-streams:{default:!a:gen-1:run-1}:chain"
+  },
+  {
+    "prefix": "p{x}",
+    "namespace": "ns:a",
+    "owner_kind": "activity",
+    "activity_id": "act:1{b}!",
+    "first_run_id": "r:1",
+    "topic": "t/1",
+    "log": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:t:t%2F1",
+    "meta": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:t:t%2F1:meta",
+    "chain": "p%7Bx%7D:{ns%3Aa:!a:act%3A1%7Bb%7D%21:r%3A1}:chain"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "ns",
+    "owner_kind": "activity",
+    "activity_id": "!a",
+    "first_run_id": "run",
+    "topic": "out",
+    "log": "temporal-streams:{ns:!a:%21a:run}:t:out",
+    "meta": "temporal-streams:{ns:!a:%21a:run}:t:out:meta",
+    "chain": "temporal-streams:{ns:!a:%21a:run}:chain"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "ns",
+    "owner_kind": "activity",
+    "activity_id": "wörk é",
+    "first_run_id": "01a1-xyz",
+    "topic": "tópic",
+    "log": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:t:t%C3%B3pic",
+    "meta": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:t:t%C3%B3pic:meta",
+    "chain": "temporal-streams:{ns:!a:w%C3%B6rk%20%C3%A9:01a1-xyz}:chain"
   }
 ]
diff --git a/crates/streams/testdata/streams/activity_attempts.json b/crates/streams/testdata/streams/activity_attempts.json
new file mode 100644
index 0000000000..bb1b9a4dc6
--- /dev/null
+++ b/crates/streams/testdata/streams/activity_attempts.json
@@ -0,0 +1,98 @@
+[
+  {
+    "name": "first attempt",
+    "attempt": 1,
+    "started_time": {
+      "seconds": 1791936000,
+      "nanos": 0
+    },
+    "key": "3669884928000001"
+  },
+  {
+    "name": "a retry starts later and keys higher",
+    "attempt": 2,
+    "started_time": {
+      "seconds": 1791936002,
+      "nanos": 500000000
+    },
+    "key": "3669884933120002"
+  },
+  {
+    "name": "the start time keeps whole milliseconds only",
+    "attempt": 3,
+    "started_time": {
+      "seconds": 1791936005,
+      "nanos": 999999
+    },
+    "key": "3669884938240003"
+  },
+  {
+    "name": "a reset attempt 1 starts later than attempt 3 and keys higher",
+    "attempt": 1,
+    "started_time": {
+      "seconds": 1791936009,
+      "nanos": 0
+    },
+    "key": "3669884946432001"
+  },
+  {
+    "name": "an attempt past 2047 keys as 2047",
+    "attempt": 3000,
+    "started_time": {
+      "seconds": 1791936010,
+      "nanos": 1000000
+    },
+    "key": "3669884948484095"
+  },
+  {
+    "name": "no start time writes the Temporal attempt",
+    "attempt": 4,
+    "started_time": null,
+    "key": "4"
+  },
+  {
+    "name": "an attempt below one is refused",
+    "attempt": 0,
+    "started_time": {
+      "seconds": 1791936000,
+      "nanos": 0
+    },
+    "key": null
+  },
+  {
+    "name": "a start time at the epoch is refused",
+    "attempt": 1,
+    "started_time": {
+      "seconds": 0,
+      "nanos": 0
+    },
+    "key": null
+  },
+  {
+    "name": "a start time before the epoch is refused",
+    "attempt": 1,
+    "started_time": {
+      "seconds": -1,
+      "nanos": 0
+    },
+    "key": null
+  },
+  {
+    "name": "a start time at 2^42 ms, in May 2109, is refused",
+    "attempt": 1,
+    "started_time": {
+      "seconds": 4398046511,
+      "nanos": 104000000
+    },
+    "key": null
+  },
+  {
+    "name": "the last millisecond before 2^42 ms keys below 2^53",
+    "attempt": 2047,
+    "started_time": {
+      "seconds": 4398046511,
+      "nanos": 103000000
+    },
+    "key": "9007199254740991"
+  }
+]
diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs
index 6f2d5612f3..f5855bdd41 100644
--- a/crates/streams/tests/conformance.rs
+++ b/crates/streams/tests/conformance.rs
@@ -17,8 +17,8 @@ use temporalio_common::protos::temporal::api::common::v1::Payload;
 use temporalio_streams::{
     BEGINNING, CHAIN_LOG, Capabilities, END, MemoryStore, NotificationEntry, NotificationLog,
     Notifier, NotifierClient, NotifyError, NotifyingStore, ReadTarget, StreamNotification,
-    StreamResult, StreamStore, append_batch, close_chain, close_topic, mint_cursor, proto::*,
-    read_page, stored_append_record, stream_hash,
+    StreamResult, StreamStore, append_batch, chain_owner, close_chain, close_topic, mint_cursor,
+    proto::*, read_page, stored_append_record, stream_hash,
 };
 
 #[async_trait::async_trait]
@@ -59,6 +59,7 @@ impl Case for Memory {
             namespace: "default".to_string(),
             workflow_id: unique("conformance"),
             first_run_id: unique("run"),
+            ..Default::default()
         }
     }
 
@@ -126,6 +127,47 @@ impl<C: Case> Case for Notifying<C> {
     }
 }
 
+/// Another case whose chains are standalone Activity runs, so every case runs on the Activity
+/// keys too.
+struct Activity<C>(C);
+
+async fn activity<C: Case>(
+    inner: impl std::future::Future<Output = Option<C>>,
+) -> Option<Activity<C>> {
+    Some(Activity(inner.await?))
+}
+
+/// The chain of standalone Activity run `run_id` of `activity_id`.
+fn activity_chain(namespace: &str, activity_id: &str, run_id: &str) -> ChainId {
+    ChainId {
+        namespace: namespace.to_string(),
+        first_run_id: run_id.to_string(),
+        owner_kind: StreamOwnerKind::Activity as i32,
+        activity_id: activity_id.to_string(),
+        ..Default::default()
+    }
+}
+
+#[async_trait::async_trait]
+impl<C: Case> Case for Activity<C> {
+    fn store(&self) -> Arc<dyn StreamStore> {
+        self.0.store()
+    }
+
+    fn chain(&self) -> ChainId {
+        let chain = self.0.chain();
+        activity_chain(&chain.namespace, &chain.workflow_id, &chain.first_run_id)
+    }
+
+    async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) {
+        self.0.drop_oldest(chain, topic, keep).await;
+    }
+
+    async fn expire_log(&self, chain: &ChainId, topic: &str) {
+        self.0.expire_log(chain, topic).await;
+    }
+}
+
 /// The memory store with only the capabilities `mask` leaves, so each fallback runs.
 async fn masked(mask: impl FnOnce(&mut Capabilities)) -> Option<Memory> {
     let mut capabilities = MemoryStore::new().capabilities();
@@ -186,11 +228,15 @@ mod on_redis {
     }
 
     pub(crate) fn log_key(prefix: &str, chain: &ChainId, topic: &str) -> String {
+        let owner = if chain.owner_kind == StreamOwnerKind::Activity as i32 {
+            format!("!a:{}", part(&chain.activity_id))
+        } else {
+            part(&chain.workflow_id)
+        };
         format!(
-            "{}:{{{}:{}:{}}}:t:{}",
+            "{}:{{{}:{owner}:{}}}:t:{}",
             part(prefix),
             part(&chain.namespace),
-            part(&chain.workflow_id),
             part(&chain.first_run_id),
             part(topic)
         )
@@ -404,6 +450,7 @@ mod on_redis {
                 namespace: "default".to_string(),
                 workflow_id: unique("conformance"),
                 first_run_id: unique("run"),
+                ..Default::default()
             }
         }
 
@@ -542,7 +589,10 @@ fn target(chain: &ChainId, topic: &str) -> ReadTarget {
     ReadTarget {
         chain: chain.clone(),
         topic: topic.to_string(),
-        stream_hash: stream_hash(&chain.namespace, "workflow", &chain.workflow_id, topic),
+        stream_hash: {
+            let (kind, id) = chain_owner(chain);
+            stream_hash(&chain.namespace, kind, id, topic)
+        },
     }
 }
 
@@ -1455,17 +1505,40 @@ mod cases {
             ..chain.clone()
         };
         let bystander = case.chain();
+        let (kind_name, id) = chain_owner(&chain);
+        let activity = kind_name == "activity";
+        // An owner of the other kind with the same id keeps its streams.
+        let twin = if activity {
+            ChainId {
+                namespace: chain.namespace.clone(),
+                workflow_id: id.to_string(),
+                first_run_id: chain.first_run_id.clone(),
+                ..Default::default()
+            }
+        } else {
+            activity_chain(&chain.namespace, id, &chain.first_run_id)
+        };
         let landed = Producer::new(case, &chain, OUT, "p", 1)
             .append(&["1"])
             .await;
         Producer::new(case, &next, OUT, "p", 1).append(&["1"]).await;
-        Producer::new(case, &bystander, OUT, "p", 1)
-            .append(&["1"])
-            .await;
-        let request = DeleteOwnerRequest {
-            namespace: chain.namespace.clone(),
-            owner_kind: StreamOwnerKind::Workflow as i32,
-            workflow_id: chain.workflow_id.clone(),
+        for kept in [&bystander, &twin] {
+            Producer::new(case, kept, OUT, "p", 1).append(&["1"]).await;
+        }
+        let request = if activity {
+            DeleteOwnerRequest {
+                namespace: chain.namespace.clone(),
+                owner_kind: StreamOwnerKind::Activity as i32,
+                activity_id: id.to_string(),
+                ..Default::default()
+            }
+        } else {
+            DeleteOwnerRequest {
+                namespace: chain.namespace.clone(),
+                owner_kind: StreamOwnerKind::Workflow as i32,
+                workflow_id: id.to_string(),
+                ..Default::default()
+            }
         };
         assert!(store.delete_owner(request.clone()).await.unwrap().deleted > 0);
         assert_eq!(store.delete_owner(request).await.unwrap().deleted, 0);
@@ -1476,10 +1549,12 @@ mod cases {
             kind(read(case, &chain, OUT, &after).await),
             StreamFailureKind::NotFound
         );
-        assert_eq!(
-            values(&read(case, &bystander, OUT, BEGINNING).await.unwrap()),
-            ["1"]
-        );
+        for kept in [&bystander, &twin] {
+            assert_eq!(
+                values(&read(case, kept, OUT, BEGINNING).await.unwrap()),
+                ["1"]
+            );
+        }
     }
 
     /// The store's log. A case runs only on a store that keeps one.
@@ -1627,8 +1702,13 @@ mod cases {
         let delete = store
             .delete_owner(DeleteOwnerRequest {
                 namespace: chain.namespace.clone(),
-                owner_kind: StreamOwnerKind::Workflow as i32,
+                owner_kind: if chain.owner_kind == StreamOwnerKind::Activity as i32 {
+                    StreamOwnerKind::Activity as i32
+                } else {
+                    StreamOwnerKind::Workflow as i32
+                },
                 workflow_id: chain.workflow_id.clone(),
+                activity_id: chain.activity_id.clone(),
             })
             .await;
         assert_eq!(unsupported(delete.map(drop)), !declared.delete_owner);
@@ -1818,6 +1898,12 @@ mod memory_with_log {
     log_cases!(super::memory_with_log());
 }
 
+mod activity_memory {
+    store_cases!(super::activity(super::memory()));
+    log_cases!(super::activity(super::memory_with_log()));
+    conformance!(super::activity(super::memory()); deleting_an_owner_drops_every_chain_of_it);
+}
+
 mod notifying_memory {
     store_cases!(super::notifying(super::memory()));
     log_cases!(super::notifying(super::memory_with_log()));
@@ -1843,6 +1929,53 @@ mod cluster_redis {
     );
 }
 
+#[cfg(feature = "redis")]
+mod activity_standalone_redis {
+    store_cases!(super::activity(super::on_redis::redis(
+        "STREAMS_REDIS_URL",
+        false
+    )));
+    log_cases!(super::activity(super::on_redis::redis_with_log(
+        "STREAMS_REDIS_URL",
+        false
+    )));
+    conformance!(super::activity(super::on_redis::redis("STREAMS_REDIS_URL", false));
+        deleting_an_owner_drops_every_chain_of_it,
+    );
+}
+
+#[cfg(feature = "redis")]
+mod activity_documented_acl_redis {
+    // The documented key rule covers an Activity's keys, since they start with the namespace.
+    store_cases!(super::activity(super::on_redis::redis_as_documented_user(
+        "STREAMS_REDIS_URL",
+        false,
+        &[]
+    )));
+    conformance!(super::activity(super::on_redis::redis_as_documented_user(
+        "STREAMS_REDIS_URL",
+        false,
+        &["+scan"]
+    ));
+        deleting_an_owner_drops_every_chain_of_it,
+    );
+}
+
+#[cfg(feature = "redis")]
+mod activity_cluster_redis {
+    store_cases!(super::activity(super::on_redis::redis(
+        "STREAMS_REDIS_CLUSTER_URL",
+        true
+    )));
+    log_cases!(super::activity(super::on_redis::redis_with_log(
+        "STREAMS_REDIS_CLUSTER_URL",
+        true
+    )));
+    conformance!(super::activity(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true));
+        deleting_an_owner_drops_every_chain_of_it,
+    );
+}
+
 #[cfg(feature = "redis")]
 mod standalone_redis_with_log {
     store_cases!(super::on_redis::redis_with_log("STREAMS_REDIS_URL", false));
diff --git a/crates/streams/tests/cross_impl.rs b/crates/streams/tests/cross_impl.rs
index 70585ff96f..9f2900a4c8 100644
--- a/crates/streams/tests/cross_impl.rs
+++ b/crates/streams/tests/cross_impl.rs
@@ -18,12 +18,10 @@ use std::{
     },
     time::Duration,
 };
-use temporalio_common::protos::temporal::api::{
-    common::v1::Payload, enums::v1::WorkflowExecutionStatus,
-};
+use temporalio_common::protos::temporal::api::common::v1::Payload;
 use temporalio_streams::{
-    OwnerClient, OwnerDescription, OwnerError, ReadTarget, RedisStore, RedisStoreOptions,
-    StreamStore, Streams, StreamsOptions, mint_cursor,
+    OwnerClient, OwnerDescription, OwnerError, OwnerId, OwnerStatus, ReadTarget, RedisStore,
+    RedisStoreOptions, StreamStore, Streams, StreamsOptions, mint_cursor,
     proto::{
         AppendRecord, AppendRequest, ChainId, NamedProducer, ReadRequest, StoreAppendRequest,
         StreamAddress, StreamFailureKind, StreamOwnerKind, StreamRecordKind,
@@ -104,6 +102,7 @@ impl Both {
             namespace: NAMESPACE.to_string(),
             workflow_id: self.workflow_id.clone(),
             first_run_id: FIRST_RUN.to_string(),
+            ..Default::default()
         }
     }
 
@@ -271,13 +270,13 @@ impl OwnerClient for Running {
     async fn describe(
         &self,
         _namespace: &str,
-        _workflow_id: &str,
+        _owner: OwnerId<'_>,
         _run_id: &str,
     ) -> Result<OwnerDescription, OwnerError> {
         Ok(OwnerDescription {
             run_id: FIRST_RUN.to_string(),
             first_run_id: FIRST_RUN.to_string(),
-            status: WorkflowExecutionStatus::Running,
+            status: OwnerStatus::Running,
         })
     }
 }
@@ -303,6 +302,7 @@ impl Both {
                 workflow_id: self.workflow_id.clone(),
                 run_id: String::new(),
                 topic: "events".to_string(),
+                ..Default::default()
             }),
             producer: Some(Producer::Named(NamedProducer {
                 producer_id: "p".to_string(),
diff --git a/crates/streams/tests/driver.rs b/crates/streams/tests/driver.rs
index c6d06d070c..21c731f90c 100644
--- a/crates/streams/tests/driver.rs
+++ b/crates/streams/tests/driver.rs
@@ -16,8 +16,9 @@ use std::{
 };
 use temporalio_streams::{
     NotifierClient, NotifyError, RedisStore, RedisStoreOptions, StreamDriver, StreamDriverOptions,
-    StreamNotification, StreamStore, progress_counter,
-    proto::{ChainId, StoreAppendRequest},
+    StreamNotification, StreamStore, chain_owner, mint_cursor, progress_counter,
+    proto::{ChainId, StoreAppendRequest, StreamOwnerKind},
+    stream_hash,
 };
 
 static IDS: AtomicUsize = AtomicUsize::new(0);
@@ -34,10 +35,12 @@ fn unique(prefix: &str) -> String {
     )
 }
 
-/// What a notifier got: each stream's highest counter, and every call.
+/// What a notifier got: each stream's highest counter and its position, by owner id and topic,
+/// and every call.
 #[derive(Default)]
 struct Notifiers {
     newest: Mutex<HashMap<(String, String), i64>>,
+    positions: Mutex<HashMap<(String, String), String>>,
     calls: AtomicUsize,
     fail: AtomicBool,
 }
@@ -57,11 +60,17 @@ impl NotifierClient for Notifiers {
         }
         self.calls.fetch_add(1, Ordering::SeqCst);
         let mut newest = self.newest.lock().unwrap();
+        let mut positions = self.positions.lock().unwrap();
         for notification in notifications {
-            let counter = newest
-                .entry((notification.chain.workflow_id, notification.topic))
-                .or_default();
-            *counter = (*counter).max(notification.counter);
+            let key = (
+                chain_owner(&notification.chain).1.to_string(),
+                notification.topic,
+            );
+            let counter = newest.entry(key.clone()).or_default();
+            if notification.counter > *counter {
+                *counter = notification.counter;
+                positions.insert(key, notification.position);
+            }
         }
         Ok(())
     }
@@ -131,6 +140,7 @@ async fn write(
                     namespace: "ns".to_string(),
                     workflow_id: workflow_id.clone(),
                     first_run_id: "run-1".to_string(),
+                    ..Default::default()
                 }),
                 topic: "out".to_string(),
                 producer_id: "p".to_string(),
@@ -149,6 +159,84 @@ async fn write(
     expected
 }
 
+/// Appends one batch to the runs of `streams` standalone Activities, `act-<index>`, and answers
+/// each Activity stream's newest counter and its cursor.
+async fn write_activities(
+    store: &RedisStore,
+    streams: usize,
+) -> BTreeMap<(String, String), (i64, String)> {
+    let mut expected = BTreeMap::new();
+    for index in 0..streams {
+        let id = format!("act-{index}");
+        let append = |chain: ChainId| StoreAppendRequest {
+            chain: Some(chain),
+            topic: "out".to_string(),
+            producer_id: "p".to_string(),
+            attempt: 1,
+            sequence: 1,
+            digest: vec![1],
+            records: vec![b"record".to_vec()],
+        };
+        let landed = store
+            .append(append(ChainId {
+                namespace: "ns".to_string(),
+                first_run_id: "run-1".to_string(),
+                owner_kind: StreamOwnerKind::Activity as i32,
+                activity_id: id.clone(),
+                ..Default::default()
+            }))
+            .await
+            .unwrap();
+        let position = mint_cursor(
+            store.name(),
+            &stream_hash("ns", "activity", &id, "out"),
+            &landed.last_position,
+        );
+        expected.insert(
+            (id, "out".to_string()),
+            (progress_counter(&landed.last_position).unwrap(), position),
+        );
+    }
+    expected
+}
+
+async fn activity_streams_reach_their_notifiers(url: &str, cluster: bool) {
+    let _shared = CLUSTER.read().await;
+    let prefix = unique("driver-activity");
+    let store = store(url, cluster, &prefix).await;
+    let notifiers = Arc::new(Notifiers::default());
+    let driver =
+        Arc::new(StreamDriver::new(store.clone(), notifiers.clone(), options("d1")).unwrap());
+    let running = tokio::spawn(driver.run());
+    let expected = write_activities(&store, 100).await;
+    let counters = expected
+        .iter()
+        .map(|(stream, (counter, _))| (stream.clone(), *counter))
+        .collect();
+    notifiers.reach(&counters, Duration::from_secs(20)).await;
+    running.abort();
+    let positions = notifiers.positions.lock().unwrap().clone();
+    for (stream, (_, position)) in &expected {
+        assert_eq!(positions.get(stream), Some(position), "{stream:?}");
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn activity_streams_on_a_server_reach_their_notifiers() {
+    let Some(url) = url("STREAMS_REDIS_URL") else {
+        return;
+    };
+    activity_streams_reach_their_notifiers(&url, false).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn activity_streams_in_a_cluster_reach_their_notifiers() {
+    let Some(url) = url("STREAMS_REDIS_CLUSTER_URL") else {
+        return;
+    };
+    activity_streams_reach_their_notifiers(&url, true).await;
+}
+
 fn url(variable: &str) -> Option<String> {
     let url = std::env::var(variable).ok();
     if url.is_none() {
@@ -422,6 +510,7 @@ async fn traffic(
                     namespace: "ns".to_string(),
                     workflow_id: workflow_id.clone(),
                     first_run_id: "run-1".to_string(),
+                    ..Default::default()
                 }),
                 topic: "out".to_string(),
                 producer_id: "p".to_string(),
diff --git a/crates/streams/tests/fallbacks.rs b/crates/streams/tests/fallbacks.rs
index c2dbc586d2..bbff868341 100644
--- a/crates/streams/tests/fallbacks.rs
+++ b/crates/streams/tests/fallbacks.rs
@@ -18,8 +18,8 @@ use temporalio_common::protos::temporal::api::{
 };
 use temporalio_streams::{
     BEGINNING, CHAIN_LOG, Capabilities, MemoryStore, OwnerClient, OwnerDescription, OwnerError,
-    ReadTarget, StreamError, StreamResult, StreamStore, Streams, StreamsOptions, close_chain,
-    proto::*, read_page, stream_hash,
+    OwnerId, OwnerStatus, ReadTarget, StreamError, StreamResult, StreamStore, Streams,
+    StreamsOptions, close_chain, proto::*, read_page, stream_hash,
 };
 
 /// A store that forwards to another and injects the faults a test asks for. It also counts
@@ -160,18 +160,18 @@ impl StreamStore for Faulty {
 
 /// Temporal, with one Workflow whose only run is `run-1`.
 struct Owner {
-    status: Mutex<Result<WorkflowExecutionStatus, ()>>,
+    status: Mutex<Result<OwnerStatus, ()>>,
 }
 
 impl Owner {
     fn running() -> Arc<Self> {
         Arc::new(Self {
-            status: Mutex::new(Ok(WorkflowExecutionStatus::Running)),
+            status: Mutex::new(Ok(OwnerStatus::Running)),
         })
     }
 
     fn end(&self, status: WorkflowExecutionStatus) {
-        *self.status.lock().unwrap() = Ok(status);
+        *self.status.lock().unwrap() = Ok(OwnerStatus::of_workflow(status));
     }
 
     fn forget(&self) {
@@ -184,7 +184,7 @@ impl OwnerClient for Owner {
     async fn describe(
         &self,
         _: &str,
-        workflow_id: &str,
+        _: OwnerId<'_>,
         _: &str,
     ) -> Result<OwnerDescription, OwnerError> {
         match *self.status.lock().unwrap() {
@@ -193,7 +193,7 @@ impl OwnerClient for Owner {
                 first_run_id: "run-1".to_string(),
                 status,
             }),
-            Err(()) => Err(OwnerError::NotFound(workflow_id.to_string())),
+            Err(()) => Err(OwnerError::NotFound("gone".to_string())),
         }
     }
 }
@@ -212,6 +212,7 @@ fn new_stream() -> StreamAddress {
         ),
         run_id: String::new(),
         topic: "out".to_string(),
+        ..Default::default()
     }
 }
 
@@ -220,6 +221,7 @@ fn chain(stream: &StreamAddress) -> ChainId {
         namespace: stream.namespace.clone(),
         workflow_id: stream.workflow_id.clone(),
         first_run_id: "run-1".to_string(),
+        ..Default::default()
     }
 }
 
diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs
index 597f543903..4aac4e51b3 100644
--- a/crates/streams/tests/redis_store.rs
+++ b/crates/streams/tests/redis_store.rs
@@ -87,6 +87,7 @@ fn chain() -> ChainId {
         namespace: "ns".to_string(),
         workflow_id: unique("wf"),
         first_run_id: "run-1".to_string(),
+        ..Default::default()
     }
 }
 
@@ -319,6 +320,7 @@ async fn ids_with_separators_and_braces_keep_their_own_keys() {
         namespace: "ns".to_string(),
         workflow_id: format!("{}:{{b}}", unique("a")),
         first_run_id: "c".to_string(),
+        ..Default::default()
     };
     let (head, tail) = one.workflow_id.split_once(':').unwrap();
     let other = ChainId {
@@ -759,13 +761,13 @@ impl temporalio_streams::OwnerClient for OneOwner {
     async fn describe(
         &self,
         _namespace: &str,
-        _workflow_id: &str,
+        _owner: temporalio_streams::OwnerId<'_>,
         _run_id: &str,
     ) -> Result<temporalio_streams::OwnerDescription, temporalio_streams::OwnerError> {
         Ok(temporalio_streams::OwnerDescription {
             run_id: "run-1".to_string(),
             first_run_id: "run-1".to_string(),
-            status: *self.status.lock().unwrap(),
+            status: temporalio_streams::OwnerStatus::of_workflow(*self.status.lock().unwrap()),
         })
     }
 }
@@ -806,6 +808,7 @@ async fn a_new_producer_marks_an_ended_chain_closed_until_it_expires() {
             workflow_id: chain.workflow_id.clone(),
             run_id: String::new(),
             topic: "events".to_string(),
+            ..Default::default()
         }),
         producer: Some(Producer::Named(NamedProducer {
             producer_id: producer.to_string(),
@@ -1274,6 +1277,7 @@ fn delete_request(workflow_id: &str) -> temporalio_streams::proto::DeleteOwnerRe
         namespace: "ns".to_string(),
         owner_kind: temporalio_streams::proto::StreamOwnerKind::Workflow as i32,
         workflow_id: workflow_id.to_string(),
+        ..Default::default()
     }
 }
 
@@ -1425,6 +1429,7 @@ async fn a_delete_reaches_every_primary_of_a_cluster() {
             namespace: "ns".to_string(),
             workflow_id: workflow_id.clone(),
             first_run_id: format!("run-{run}"),
+            ..Default::default()
         };
         store
             .append(append(&chain, "p", 1, 1, b"\x01"))
@@ -1731,6 +1736,7 @@ async fn a_replay_read_on_redis_repeats_the_read_and_survives_retention_under_a_
         workflow_id: chain.workflow_id.clone(),
         run_id: String::new(),
         topic: "events".to_string(),
+        ..Default::default()
     };
     let append = |attempt: i64, sequence: i64, value: &str| AppendRequest {
         stream: Some(stream.clone()),
@@ -1817,6 +1823,7 @@ async fn a_replay_read_from_a_trimmed_start_on_redis_is_rec_01() {
         workflow_id: chain.workflow_id.clone(),
         run_id: String::new(),
         topic: "events".to_string(),
+        ..Default::default()
     };
     let append = |sequence: i64| AppendRequest {
         stream: Some(stream.clone()),
@@ -1897,6 +1904,7 @@ async fn a_chains_notification_entry_lands_in_the_chains_own_slot() {
             namespace: unique("ns:with{braces}"),
             workflow_id: unique("wf"),
             first_run_id: unique("run"),
+            ..Default::default()
         };
         let landed = store
             .append(append(&chain, "p", 1, 1, b"\x01"))
@@ -2076,6 +2084,7 @@ async fn a_replay_whose_end_record_the_store_lost_is_rec_01() {
         workflow_id: chain.workflow_id.clone(),
         run_id: String::new(),
         topic: "events".to_string(),
+        ..Default::default()
     };
     let append = |sequence: i64| AppendRequest {
         stream: Some(stream.clone()),
diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs
index 9d3eb615db..7715bb42d8 100644
--- a/crates/streams/tests/vectors.rs
+++ b/crates/streams/tests/vectors.rs
@@ -29,3 +29,36 @@ fn cursors_match_the_python_sdk() {
         }
     }
 }
+
+#[test]
+fn activity_attempt_keys_match_the_vectors() {
+    // Every SDK reports the attempt a record carries, so each one composes it the same way.
+    use temporalio_streams::{activity_attempt, proto::ActivityProducer};
+    let cases: Vec<Value> =
+        serde_json::from_str(include_str!("../testdata/streams/activity_attempts.json")).unwrap();
+    assert!(cases.iter().any(|case| case["key"].is_null()));
+    for case in &cases {
+        let mut activity = ActivityProducer {
+            activity_id: "act".to_string(),
+            run_id: "run".to_string(),
+            attempt: case["attempt"].as_i64().unwrap() as i32,
+            ..Default::default()
+        };
+        if let Some(started) = case["started_time"].as_object() {
+            let time = activity
+                .attempt_started_time
+                .get_or_insert_with(Default::default);
+            time.seconds = started["seconds"].as_i64().unwrap();
+            time.nanos = started["nanos"].as_i64().unwrap() as i32;
+        }
+        let name = case["name"].as_str().unwrap();
+        match case["key"].as_str() {
+            Some(key) => assert_eq!(
+                activity_attempt(&activity).unwrap().to_string(),
+                key,
+                "{name}"
+            ),
+            None => assert!(activity_attempt(&activity).is_err(), "{name}"),
+        }
+    }
+}