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 6c2dfa80c5..4f53e16d8b 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
@@ -22,4 +22,11 @@ enum StreamOwnerKind {
     // `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;
+    // An Activity that a Workflow scheduled owns the stream. `workflow_id` names the Workflow,
+    // `run_id` the run that scheduled the Activity, and `activity_id` the Activity. The stream
+    // closes when the Workflow sees the Activity end. If the Activity is no longer pending in the
+    // run and no close came, the stream notifier closes the stream with a failure, after
+    // `streamnotifier.ownerStartGrace` from its start and `streamnotifier.ownerCloseGrace` from
+    // the first check that found the Activity gone.
+    STREAM_OWNER_KIND_WORKFLOW_ACTIVITY = 3;
 }
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 c0a76d5989..063f5f2811 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,17 +18,20 @@ 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`. Empty for other kinds.
+    // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. The Workflow that scheduled the
+    // owning Activity, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds.
     string workflow_id = 2;
     // 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.
+    // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. For
+    // `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`, the id of the run that scheduled the Activity, not
+    // the chain's first 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.
+    // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. The owning Activity's id
+    // in the scheduling run, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds.
     string activity_id = 5;
 }
 
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 6f43676b90..7d44f87e11 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
@@ -132,29 +132,36 @@ enum StreamOwnerKind {
   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;
+  // An Activity that a Workflow run scheduled. It is a chain of one, keyed by the scheduling run
+  // and the Activity id, so its streams end with the Activity.
+  STREAM_OWNER_KIND_WORKFLOW_ACTIVITY = 3;
 }
 
 // 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.
+  // The owning Workflow, or the Workflow that scheduled a Workflow Activity. Empty for a
+  // standalone 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. For an Activity owner, the Activity run, and empty names
-  // the newest run of the Activity id when the call starts.
+  // the newest run of the Activity id when the call starts. For a Workflow Activity owner, the
+  // run that scheduled the Activity. It is required, since an Activity id names nothing across
+  // runs.
   string run_id = 4;
   string topic = 5;
-  // The owning standalone Activity. Empty for a Workflow owner.
+  // The owning 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. A standalone
-// Activity run is a chain of one run, with the run as `first_run_id`.
+// Activity run is a chain of one run, with the run as `first_run_id`. A Workflow Activity is a
+// chain of one, with the scheduling run as `first_run_id`.
 message ChainId {
   string namespace = 1;
-  // Empty for an Activity owner.
+  // Empty for a standalone 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.
@@ -448,7 +455,8 @@ message RunStreamDriverResponse {}
 message DeleteOwnerRequest {
   string namespace = 1;
   StreamOwnerKind owner_kind = 2;
-  // The Workflow whose streams go, for a Workflow owner.
+  // The Workflow whose streams go, for a Workflow owner. The streams of every Activity that the
+  // Workflow scheduled go too.
   string workflow_id = 3;
   // The standalone Activity whose streams go, across all its runs, for an Activity owner.
   string activity_id = 4;
diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md
index 6de48a00ec..7467883963 100644
--- a/crates/sdk-core/CHANGELOG.md
+++ b/crates/sdk-core/CHANGELOG.md
@@ -172,6 +172,34 @@ relevant information.
   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.
+* Experimental: the stream protos name a Workflow Activity owner. `StreamOwnerKind` gains
+  `WORKFLOW_ACTIVITY`, an Activity that a Workflow run scheduled. Its `StreamAddress` and
+  `ChainId` set `workflow_id`, the scheduling run and `activity_id`, with no new field. The
+  vendored API's owner kind gains `WORKFLOW_ACTIVITY`. Core still refuses the kind as
+  `UNSUPPORTED`.
+* Experimental: a Workflow Activity owns streams. Its `StreamAddress` names the Workflow, the
+  scheduling run and the Activity id, all required. It is a chain of one, keyed by that run, so
+  Activity `1` of the next run after Continue-as-New is another stream. Core describes the run
+  and looks up the Activity in `pending_activities`. A closed run ends reads at once. A running
+  run that no longer lists the Activity ends reads only after the owner start grace from the
+  first check that found it so, and refuses producers at once. Cursors hash under the kind
+  `workflow_activity` with the run in the owner id. The Redis keys have the hash tag
+  `{<namespace>:!wa:<workflow id>:<run id>:<activity id>}`, and the append script (hashes
+  `eec20388`, and `418b9d10` with the notification log) adds each log key to the chain's
+  `topics` set on a producer attempt's first batch. `StreamStore::written_topics` answers that
+  set, and every store and wrapper must implement it. A Workflow's `DeleteOwner` also deletes
+  its Activities' streams. `OwnerId` gains `WorkflowActivity` and `OwnerStatus` gains
+  `ActivityGone`. The documented Redis ACL gains `+sadd +smembers`.
+* Experimental: the scheduling Workflow's Core closes a Workflow Activity's streams when it
+  resolves the Activity. Workflow code marks the Activity with the `ScheduleActivity` header
+  `__temporal_stream_owner`. When a marked Activity resolves outside replay, Core closes each
+  topic in the Activity's topic set in the background, once the activation completed and its
+  stream output was appended. A completed Activity closes with a null result, and a failed,
+  timed-out or canceled one with the failure its resolution holds, cut to `MAX_CLOSE_BYTES`. The
+  close names `<activity id>@<run id>` as `closed_by` and reaches the stream's notifier. A close
+  that fails is logged and does not fail the Workflow Task. Unmarked Activities cost no store
+  call. Notifier calls name a Workflow Activity stream by its Workflow, scheduling run and
+  Activity id.
 
 ### 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 1f216fecda..f69e308fa8 100644
--- a/crates/sdk-core/src/core_tests/external_streams.rs
+++ b/crates/sdk-core/src/core_tests/external_streams.rs
@@ -3,7 +3,7 @@
 
 use crate::{
     TaskToken,
-    replay::{TestHistoryBuilder, canned_histories, default_wes_attribs},
+    replay::{TestHistoryBuilder, canned_histories, default_act_sched, default_wes_attribs},
     test_help::{
         MockPollCfg, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers,
         hist_to_poll_resp, mock_worker, schedule_local_activity_cmd, start_timer_cmd,
@@ -22,7 +22,8 @@ use temporalio_common::{
             workflow_activation::{WorkflowActivation, workflow_activation_job},
             workflow_commands::{
                 ActivityCancellationType, CompleteWorkflowExecution, OutputRecord,
-                WorkflowOutputStreamCommit, workflow_command,
+                RequestCancelActivity, ScheduleActivity, WorkflowOutputStreamCommit,
+                workflow_command,
             },
             workflow_completion::WorkflowActivationCompletion,
         },
@@ -48,7 +49,7 @@ use temporalio_streams::{
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, ReadRequest, StoreAppendRequest,
         StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest,
-        StoreReadResponse, read_record,
+        StoreReadResponse, StoredClose, StreamOwnerKind, read_record,
     },
     read_page, stream_hash, workflow_batch_digest,
 };
@@ -68,6 +69,13 @@ struct RecordingStore {
     refuse: Option<StreamError>,
     /// How long each append takes.
     append_delay: Option<Duration>,
+    /// Each stream close as `(chain, topic, close)`.
+    closes: Mutex<Vec<(ChainId, String, StoredClose)>>,
+    /// Refuses every topic close.
+    refuse_closes: bool,
+    /// How long each topic close takes.
+    close_delay: Option<Duration>,
+
     /// A retire waits for a permit here, when set.
     retire_gate: Option<Arc<tokio::sync::Semaphore>>,
 }
@@ -197,6 +205,25 @@ impl StreamStore for RecordingStore {
         self.inner.close_topic(chain, topic).await
     }
 
+    async fn close_topic_with(
+        &self,
+        chain: &ChainId,
+        topic: &str,
+        close: StoredClose,
+    ) -> StreamResult<StoredClose> {
+        self.calls.lock().push("close_topic_with");
+        if self.refuse_closes {
+            return Err(StreamError::storage("the store is down"));
+        }
+        if let Some(delay) = self.close_delay {
+            tokio::time::sleep(delay).await;
+        }
+        self.closes
+            .lock()
+            .push((chain.clone(), topic.to_string(), close.clone()));
+        self.inner.close_topic_with(chain, topic, close).await
+    }
+
     async fn retire_producer(
         &self,
         chain: &ChainId,
@@ -216,6 +243,11 @@ impl StreamStore for RecordingStore {
             .await
     }
 
+    async fn written_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
+        self.calls.lock().push("written_topics");
+        self.inner.written_topics(chain).await
+    }
+
     async fn list_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
         self.calls.lock().push("list_topics");
         self.inner.list_topics(chain).await
@@ -1536,3 +1568,347 @@ async fn a_rerun_after_the_run_closed_its_own_topic_drops_its_records_and_goes_o
         "the rerun's records were dropped"
     );
 }
+
+/// What a Workflow Activity's stream holds once the agent `fakeid` scheduled it in `run_id`.
+fn activity_chain(run_id: &str, activity_id: &str) -> ChainId {
+    ChainId {
+        namespace: "default".to_string(),
+        workflow_id: "fakeid".to_string(),
+        first_run_id: run_id.to_string(),
+        owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+        activity_id: activity_id.to_string(),
+    }
+}
+
+/// A `ScheduleActivity` that marks Activity `activity_id` as the owner of its streams, or not.
+fn schedule_activity(
+    activity_id: &str,
+    marked: bool,
+    cancellation_type: ActivityCancellationType,
+) -> workflow_command::Variant {
+    let mut schedule = ScheduleActivity {
+        seq: 1,
+        activity_id: activity_id.to_string(),
+        cancellation_type: cancellation_type as i32,
+        ..default_act_sched()
+    };
+    if marked {
+        schedule
+            .headers
+            .insert("__temporal_stream_owner".to_string(), Payload::default());
+    }
+    schedule.into()
+}
+
+/// Appends one record to each of `topics` of the Activity's stream, as its producer would.
+async fn activity_wrote(store: &RecordingStore, run_id: &str, activity_id: &str, topics: &[&str]) {
+    for topic in topics {
+        store
+            .inner
+            .append(StoreAppendRequest {
+                chain: Some(activity_chain(run_id, activity_id)),
+                topic: topic.to_string(),
+                producer_id: format!("{activity_id}@{run_id}"),
+                attempt: 1,
+                sequence: 1,
+                digest: vec![1; 32],
+                records: vec![b"record".to_vec()],
+            })
+            .await
+            .unwrap();
+    }
+}
+
+/// Waits a bounded time for `count` stream closes, which run in the background.
+async fn closes_within_deadline(
+    store: &RecordingStore,
+    count: usize,
+) -> Vec<(String, StoredClose)> {
+    for _ in 0..200 {
+        let closes = store.closes.lock().clone();
+        if closes.len() >= count {
+            return closes
+                .into_iter()
+                .map(|(chain, topic, close)| {
+                    assert_eq!(chain.owner_kind, StreamOwnerKind::WorkflowActivity as i32);
+                    (topic, close)
+                })
+                .collect();
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!(
+        "expected {count} stream closes, got {:?}",
+        store.closes.lock()
+    );
+}
+
+/// Runs a Workflow that schedules Activity `tool`, which writes `topics`, and completes once the
+/// Activity resolves. Returns the run id.
+async fn run_activity_owner(
+    history: TestHistoryBuilder,
+    store: Arc<RecordingStore>,
+    marked: bool,
+    topics: &[&str],
+) -> String {
+    let worker = worker_appending_to(
+        history,
+        vec![1.into(), 2.into()],
+        store.clone(),
+        Default::default(),
+    );
+    let start = worker.poll_workflow_activation().await.unwrap();
+    let run_id = start.run_id.clone();
+    activity_wrote(&store, &run_id, "tool", topics).await;
+    complete(
+        &worker,
+        &run_id,
+        vec![schedule_activity(
+            "tool",
+            marked,
+            ActivityCancellationType::TryCancel,
+        )],
+    )
+    .await;
+    let resolved = worker.poll_workflow_activation().await.unwrap();
+    assert!(!resolved.is_replaying);
+    assert!(resolved.jobs.iter().any(|job| matches!(
+        job.variant,
+        Some(workflow_activation_job::Variant::ResolveActivity(_))
+    )));
+    worker.complete_execution(&resolved.run_id).await;
+    worker.drain_pollers_and_shutdown().await;
+    run_id
+}
+
+#[tokio::test]
+async fn a_marked_activity_that_completes_closes_each_topic_it_wrote_with_null() {
+    let store = Arc::new(RecordingStore::default());
+    let run_id = run_activity_owner(
+        canned_histories::single_activity("tool"),
+        store.clone(),
+        true,
+        &["status", "tokens"],
+    )
+    .await;
+    let closes = closes_within_deadline(&store, 2).await;
+    assert_eq!(
+        closes
+            .iter()
+            .map(|(topic, _)| topic.as_str())
+            .collect::<Vec<_>>(),
+        ["status", "tokens"]
+    );
+    for (_, close) in &closes {
+        assert_eq!(close.result, Some(temporalio_streams::empty_close_result()));
+        assert!(close.failure.is_none());
+        assert_eq!(close.closed_by, format!("tool@{run_id}"));
+    }
+    let chain = store.closes.lock()[0].0.clone();
+    assert_eq!(chain, activity_chain(&run_id, "tool"));
+    // A reader of the stream finds the close.
+    let read = store
+        .inner
+        .read(StoreReadRequest {
+            chain: Some(chain),
+            topic: "tokens".to_string(),
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    assert!(read.closed);
+}
+
+#[tokio::test]
+async fn a_worker_shutdown_waits_for_the_activity_stream_closes() {
+    let store = Arc::new(RecordingStore {
+        close_delay: Some(Duration::from_millis(200)),
+        ..Default::default()
+    });
+    run_activity_owner(
+        canned_histories::single_activity("tool"),
+        store.clone(),
+        true,
+        &["tokens"],
+    )
+    .await;
+    assert_eq!(
+        store.closes.lock().len(),
+        1,
+        "closed before the shutdown returned"
+    );
+}
+
+#[tokio::test]
+async fn a_marked_activity_that_fails_closes_with_the_failure_the_workflow_saw() {
+    let store = Arc::new(RecordingStore::default());
+    run_activity_owner(
+        canned_histories::single_failed_activity("tool"),
+        store.clone(),
+        true,
+        &["tokens"],
+    )
+    .await;
+    let closes = closes_within_deadline(&store, 1).await;
+    let failure = closes[0].1.failure.clone().unwrap();
+    assert!(closes[0].1.result.is_none());
+    assert!(
+        matches!(
+            failure.failure_info,
+            Some(temporalio_common::protos::temporal::api::failure::v1::failure::FailureInfo::ActivityFailureInfo(_))
+        ),
+        "{failure:?}"
+    );
+}
+
+#[tokio::test]
+async fn a_marked_activity_that_times_out_closes_with_its_timeout() {
+    let store = Arc::new(RecordingStore::default());
+    run_activity_owner(
+        canned_histories::started_activity_timeout("tool"),
+        store.clone(),
+        true,
+        &["tokens"],
+    )
+    .await;
+    let closes = closes_within_deadline(&store, 1).await;
+    let failure = closes[0].1.failure.clone().unwrap();
+    assert_eq!(failure.message, "Activity task timed out", "{failure:?}");
+}
+
+#[tokio::test]
+async fn an_unmarked_activity_costs_no_store_call() {
+    let store = Arc::new(RecordingStore::default());
+    run_activity_owner(
+        canned_histories::single_activity("tool"),
+        store.clone(),
+        false,
+        &[],
+    )
+    .await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    // Completing the run closes its own chain, and nothing else reaches the store.
+    assert_eq!(store.calls(), ["close_chain"]);
+}
+
+#[tokio::test]
+async fn a_store_that_refuses_the_close_leaves_the_task_completed() {
+    let store = Arc::new(RecordingStore {
+        refuse_closes: true,
+        ..Default::default()
+    });
+    let completions: Arc<Mutex<Vec<usize>>> = Default::default();
+    let worker = worker_appending_to(
+        canned_histories::single_activity("tool"),
+        vec![1.into(), 2.into()],
+        store.clone(),
+        completions.clone(),
+    );
+    let start = worker.poll_workflow_activation().await.unwrap();
+    activity_wrote(&store, &start.run_id, "tool", &["tokens"]).await;
+    complete(
+        &worker,
+        &start.run_id,
+        vec![schedule_activity(
+            "tool",
+            true,
+            ActivityCancellationType::TryCancel,
+        )],
+    )
+    .await;
+    let resolved = worker.poll_workflow_activation().await.unwrap();
+    worker.complete_execution(&resolved.run_id).await;
+    worker.drain_pollers_and_shutdown().await;
+    for _ in 0..200 {
+        if store.calls().contains(&"close_topic_with") {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    assert!(
+        store.calls().contains(&"close_topic_with"),
+        "{:?}",
+        store.calls()
+    );
+    assert_eq!(completions.lock().len(), 2, "both tasks completed");
+}
+
+#[tokio::test]
+async fn replay_of_a_resolution_makes_no_close() {
+    let mut history = canned_histories::single_activity("tool");
+    // Another task after the resolution, so the resolution replays.
+    history.add_workflow_task_completed();
+    let timer_started = history.add_by_type(EventType::TimerStarted);
+    history.add_timer_fired(timer_started, "1".to_string());
+    history.add_workflow_task_scheduled_and_started();
+    let store = Arc::new(RecordingStore::default());
+    let worker = worker_appending_to(history, vec![3.into()], store.clone(), Default::default());
+    let start = worker.poll_workflow_activation().await.unwrap();
+    assert!(start.is_replaying);
+    complete(
+        &worker,
+        &start.run_id,
+        vec![schedule_activity(
+            "tool",
+            true,
+            ActivityCancellationType::TryCancel,
+        )],
+    )
+    .await;
+    let resolved = worker.poll_workflow_activation().await.unwrap();
+    assert!(resolved.is_replaying);
+    complete(
+        &worker,
+        &resolved.run_id,
+        vec![start_timer_cmd(1, Duration::from_secs(1))],
+    )
+    .await;
+    let fired = worker.poll_workflow_activation().await.unwrap();
+    worker.complete_execution(&fired.run_id).await;
+    worker.drain_pollers_and_shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    // Completing the run closes its own chain, and nothing else reaches the store.
+    assert_eq!(store.calls(), ["close_chain"]);
+}
+
+#[tokio::test]
+async fn a_try_cancel_closes_the_streams_at_once_with_the_canceled_failure() {
+    let store = Arc::new(RecordingStore::default());
+    let worker = worker_appending_to(
+        canned_histories::cancel_scheduled_activity("tool", "signal"),
+        vec![1.into(), 2.into()],
+        store.clone(),
+        Default::default(),
+    );
+    let start = worker.poll_workflow_activation().await.unwrap();
+    activity_wrote(&store, &start.run_id, "tool", &["tokens"]).await;
+    complete(
+        &worker,
+        &start.run_id,
+        vec![schedule_activity(
+            "tool",
+            true,
+            ActivityCancellationType::TryCancel,
+        )],
+    )
+    .await;
+    let signaled = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &signaled.run_id,
+        vec![RequestCancelActivity { seq: 1 }.into()],
+    )
+    .await;
+    let closes = closes_within_deadline(&store, 1).await;
+    let failure = closes[0].1.failure.clone().unwrap();
+    assert!(
+        matches!(
+            failure.cause.as_deref().and_then(|cause| cause.failure_info.as_ref()),
+            Some(temporalio_common::protos::temporal::api::failure::v1::failure::FailureInfo::CanceledFailureInfo(_))
+        ),
+        "{failure:?}"
+    );
+    let resolved = worker.poll_workflow_activation().await.unwrap();
+    worker.complete_execution(&resolved.run_id).await;
+    worker.drain_pollers_and_shutdown().await;
+}
diff --git a/crates/sdk-core/src/streams.rs b/crates/sdk-core/src/streams.rs
index 80c98a7a6b..af700d8b89 100644
--- a/crates/sdk-core/src/streams.rs
+++ b/crates/sdk-core/src/streams.rs
@@ -170,8 +170,18 @@ 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. An Activity run is its own chain, so its notifier is keyed by the run.
+/// its own. An Activity run is its own chain, so its notifier is keyed by the run. A Workflow
+/// Activity's notifier is keyed by its Workflow, its scheduling run and its Activity id.
 fn chain_reference(chain: &ChainId, topic: String) -> StreamReference {
+    if chain.owner_kind == proto::StreamOwnerKind::WorkflowActivity as i32 {
+        return StreamReference {
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            workflow_id: chain.workflow_id.clone(),
+            run_id: chain.first_run_id.clone(),
+            topic,
+            activity_id: chain.activity_id.clone(),
+        };
+    }
     if chain.owner_kind == proto::StreamOwnerKind::Activity as i32 {
         return StreamReference {
             owner_kind: StreamOwnerKind::Activity as i32,
@@ -331,6 +341,27 @@ impl OwnerClient for ConnectionOwnerClient {
                     .map_err(owner_error)?;
                 activity_owner_description(response.into_inner())
             }
+            OwnerId::WorkflowActivity {
+                workflow_id,
+                activity_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)?;
+                workflow_activity_owner_description(response.into_inner(), activity_id)
+            }
         }
     }
 }
@@ -379,6 +410,26 @@ fn owner_description(
     })
 }
 
+/// A Workflow Activity is pending while its scheduling run lists it, through retries and backoff.
+fn workflow_activity_owner_description(
+    response: DescribeWorkflowExecutionResponse,
+    activity_id: &str,
+) -> Result<OwnerDescription, OwnerError> {
+    let pending = response
+        .pending_activities
+        .iter()
+        .any(|activity| activity.activity_id == activity_id);
+    let status = response
+        .workflow_execution_info
+        .as_ref()
+        .map(|info| info.status())
+        .unwrap_or_default();
+    Ok(OwnerDescription {
+        status: OwnerStatus::of_workflow_activity(status, pending),
+        ..owner_description(response)?
+    })
+}
+
 /// A standalone Activity run is a chain of one run, so it is its own first run.
 fn activity_owner_description(
     response: DescribeActivityExecutionResponse,
@@ -480,6 +531,37 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn a_workflow_activity_runs_while_its_running_run_lists_it_as_pending() {
+        use temporalio_common::protos::temporal::api::workflow::v1::PendingActivityInfo;
+        let described = |status: WorkflowExecutionStatus, pending: &[&str]| {
+            let mut response = described("run-1", "run-1");
+            if let Some(info) = &mut response.workflow_execution_info {
+                info.status = status as i32;
+            }
+            response.pending_activities = pending
+                .iter()
+                .map(|id| PendingActivityInfo {
+                    activity_id: id.to_string(),
+                    ..Default::default()
+                })
+                .collect();
+            response
+        };
+        let status = |status, pending: &[&str]| {
+            workflow_activity_owner_description(described(status, pending), "tool")
+                .unwrap()
+                .status
+        };
+        use WorkflowExecutionStatus::*;
+        assert_eq!(status(Running, &["other", "tool"]), OwnerStatus::Running);
+        assert_eq!(status(Paused, &["tool"]), OwnerStatus::Running);
+        assert_eq!(status(Running, &["other"]), OwnerStatus::ActivityGone);
+        for closed in [Completed, Failed, ContinuedAsNew, Terminated, TimedOut] {
+            assert_eq!(status(closed, &["tool"]), OwnerStatus::Ended, "{closed:?}");
+        }
+    }
+
     #[test]
     fn a_run_without_a_first_run_id_starts_its_chain() {
         assert_eq!(
@@ -556,6 +638,33 @@ mod tests {
         assert!(request.close_result.is_none());
     }
 
+    #[test]
+    fn a_notification_of_a_workflow_activity_stream_names_its_scheduling_run() {
+        let request = notify_request(StreamNotification {
+            chain: proto::ChainId {
+                namespace: "ns".to_string(),
+                workflow_id: "agent".to_string(),
+                first_run_id: "run-2".to_string(),
+                owner_kind: proto::StreamOwnerKind::WorkflowActivity as i32,
+                activity_id: "tool-42".to_string(),
+            },
+            topic: "tokens".to_string(),
+            position: "cursor".to_string(),
+            counter: 3,
+            close: None,
+        });
+        assert_eq!(
+            request.stream_ref,
+            Some(StreamReference {
+                owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+                workflow_id: "agent".to_string(),
+                run_id: "run-2".to_string(),
+                topic: "tokens".to_string(),
+                activity_id: "tool-42".to_string(),
+            })
+        );
+    }
+
     #[test]
     fn a_notification_of_an_activity_stream_names_the_activity_run() {
         let request = notify_request(StreamNotification {
diff --git a/crates/sdk-core/src/worker/workflow/machines/activity_state_machine.rs b/crates/sdk-core/src/worker/workflow/machines/activity_state_machine.rs
index 0d9a2dc5f4..3287386bc8 100644
--- a/crates/sdk-core/src/worker/workflow/machines/activity_state_machine.rs
+++ b/crates/sdk-core/src/worker/workflow/machines/activity_state_machine.rs
@@ -9,7 +9,7 @@ use crate::{
     internal_flags::CoreInternalFlags,
     worker::workflow::{
         CommandAnnotations, InternalFlagsRef, ProtoCommandExt, fatal, machines::HistEventData,
-        nondeterminism,
+        nondeterminism, stream_output::STREAM_OWNER_HEADER,
     },
 };
 use std::convert::{TryFrom, TryInto};
@@ -224,6 +224,15 @@ impl ActivityMachine {
     pub(super) fn was_cancelled_before_sent_to_server(&self) -> bool {
         self.shared_state().cancelled_before_sent
     }
+
+    /// The Activity's id when Workflow code marked it as the owner of its own streams.
+    pub(super) fn stream_owner_activity_id(&self) -> Option<&str> {
+        let attrs = &self.shared_state.attrs;
+        attrs
+            .headers
+            .contains_key(STREAM_OWNER_HEADER)
+            .then_some(attrs.activity_id.as_str())
+    }
 }
 
 impl TryFrom<HistEventData> for ActivityMachineEvents {
diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
index b21508219b..8ab7e8543d 100644
--- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
+++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs
@@ -37,6 +37,7 @@ use crate::{
                 upsert_search_attributes_state_machine::upsert_search_attrs_internal,
             },
             nondeterminism,
+            stream_output::ActivityStreamClose,
         },
     },
 };
@@ -151,6 +152,9 @@ pub(crate) struct WorkflowMachines {
     pub(crate) last_processed_event: i64,
     /// True if the workflow is replaying from history
     pub(crate) replaying: bool,
+    /// The streams of Workflow Activities this run resolved live, to close once the activation
+    /// completes.
+    activity_stream_closes: Vec<ActivityStreamClose>,
     /// Workflow identifier
     pub(crate) workflow_id: String,
     /// Workflow type identifier. (Function name, class, etc)
@@ -327,6 +331,7 @@ impl WorkflowMachines {
             last_history_from_server: basics.history,
             protocol_msgs: vec![],
             pending_stream_wakes: vec![],
+            activity_stream_closes: vec![],
             workflow_id: basics.workflow_id,
             workflow_type: basics.workflow_type,
             run_id: basics.run_id,
@@ -368,6 +373,11 @@ impl WorkflowMachines {
         }
     }
 
+    /// Takes the stream closes of the Workflow Activities resolved since the last call.
+    pub(crate) fn take_activity_stream_closes(&mut self) -> Vec<ActivityStreamClose> {
+        std::mem::take(&mut self.activity_stream_closes)
+    }
+
     /// Returns true if workflow has seen a terminal command
     pub(crate) const fn workflow_is_finished(&self) -> bool {
         self.workflow_end_time.is_some()
@@ -1302,6 +1312,22 @@ impl WorkflowMachines {
         for response in machine_responses {
             match response {
                 MachineResponse::PushWFJob(a) => {
+                    // Replay does no store calls, so only a live resolution closes the streams.
+                    if !self.replaying
+                        && let workflow_activation_job::Variant::ResolveActivity(resolved) =
+                            &a.variant
+                        && !resolved.is_local
+                        && let Machines::ActivityMachine(machine) = self.machine(smk)
+                        && let Some(activity_id) = machine.stream_owner_activity_id()
+                        && let Some(close) = ActivityStreamClose::of(
+                            &self.workflow_id,
+                            &self.run_id,
+                            activity_id,
+                            resolved.result.as_ref(),
+                        )
+                    {
+                        self.activity_stream_closes.push(close);
+                    }
                     // We don't need to notify lang about jobs created by core-internal machines
                     if !self.machine_is_core_created.contains_key(smk) {
                         self.drive_me.send_job(a);
diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs
index 33f91b6d8a..5a5b0e99c3 100644
--- a/crates/sdk-core/src/worker/workflow/managed_run.rs
+++ b/crates/sdk-core/src/worker/workflow/managed_run.rs
@@ -907,6 +907,7 @@ impl ManagedRun {
                             replaying: self.wfm.machines.replaying,
                             output: None,
                             retired: None,
+                            activity_stream_closes: vec![],
                         },
                         resp_chan: completion.resp_chan,
                     }))
@@ -917,10 +918,12 @@ impl ManagedRun {
                 complete_resp_chan: completion.resp_chan,
             }),
         };
-        // Only a completion the run took has output to append.
+        // Only a completion the run took has output to append and streams to close.
+        let activity_stream_closes = self.wfm.machines.take_activity_stream_closes();
         completed.map(|done| {
             done.map(|mut done| {
                 done.result.output = output;
+                done.result.activity_stream_closes = activity_stream_closes;
                 done
             })
         })
@@ -1460,6 +1463,7 @@ impl ManagedRun {
                 replaying,
                 output: None,
                 retired,
+                activity_stream_closes: vec![],
             },
             resp_chan,
         }
@@ -1496,6 +1500,7 @@ impl ManagedRun {
                     replaying: self.wfm.machines.replaying,
                     output: None,
                     retired: None,
+                    activity_stream_closes: vec![],
                 })
                 .is_err()
         {
diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs
index 460e25dc43..eee8844a6d 100644
--- a/crates/sdk-core/src/worker/workflow/mod.rs
+++ b/crates/sdk-core/src/worker/workflow/mod.rs
@@ -638,7 +638,7 @@ impl Workflows {
             );
         }
 
-        let completion_outcome = if let Ok(c) = rx.await {
+        let mut completion_outcome = if let Ok(c) = rx.await {
             c
         } else if is_empty_completion {
             // The stream may shut down after accepting the message but before processing
@@ -654,8 +654,12 @@ impl Workflows {
             );
             return Ok(());
         };
+        let activity_stream_closes = std::mem::take(&mut completion_outcome.activity_stream_closes);
         let (completion_outcome, output_appended) =
             self.append_output(&run_id, completion_outcome).await;
+        if output_appended {
+            self.close_activity_streams(activity_stream_closes);
+        }
         let replaying = completion_outcome.replaying;
 
         let mut wft_from_complete = None;
@@ -765,6 +769,18 @@ impl Workflows {
         ))
     }
 
+    /// Closes the streams of the Workflow Activities a completion resolved, in the background.
+    fn close_activity_streams(&self, closes: Vec<stream_output::ActivityStreamClose>) {
+        #[cfg(feature = "streams")]
+        if let Some(store) = &self.output_store {
+            store.close_activity_streams(closes);
+            return;
+        }
+        if !closes.is_empty() {
+            debug!("This Worker has no stream store to close Workflow Activity streams in");
+        }
+    }
+
     fn retire_stream_sessions(&self, retired: stream_output::RetiredSessions) {
         #[cfg(feature = "streams")]
         if let Some(store) = &self.output_store {
@@ -1412,6 +1428,8 @@ struct ActivationCompleteResult {
     output: Option<Box<ActivationOutput>>,
     /// The store sessions to retire once the server accepts the reported task.
     retired: Option<Box<stream_output::RetiredSessions>>,
+    /// The streams of Workflow Activities resolved live, to close once the output is appended.
+    activity_stream_closes: Vec<stream_output::ActivityStreamClose>,
 }
 
 /// One activation's stream output, and what failing its Workflow Task takes.
diff --git a/crates/sdk-core/src/worker/workflow/stream_output.rs b/crates/sdk-core/src/worker/workflow/stream_output.rs
index 21c49351ea..1aaa33f704 100644
--- a/crates/sdk-core/src/worker/workflow/stream_output.rs
+++ b/crates/sdk-core/src/worker/workflow/stream_output.rs
@@ -5,13 +5,76 @@ use super::WFMachinesError;
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 use temporalio_common::{
     protos::{
-        coresdk::workflow_commands::OutputRecord, temporal::sdk::streams::v1::StreamRecordKind,
+        coresdk::{
+            activity_result::{ActivityResolution, activity_resolution},
+            workflow_commands::OutputRecord,
+        },
+        temporal::{api::failure::v1::Failure, sdk::streams::v1::StreamRecordKind},
     },
     streams::check_topic,
 };
 
 const CONTENT_HASH_BYTES: usize = 32;
 
+/// The `ScheduleActivity` header that marks an Activity as the owner of its own streams. The
+/// scheduling Workflow's Core closes them when it resolves the Activity. The command holds it, so
+/// replay rebuilds the mark with no History read, and the Activity's Worker sees it too.
+pub(crate) const STREAM_OWNER_HEADER: &str = "__temporal_stream_owner";
+
+/// The streams of one Workflow Activity to close, with the outcome its scheduling Workflow saw.
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct ActivityStreamClose {
+    pub(crate) workflow_id: String,
+    /// The run that scheduled the Activity, which keys its streams.
+    pub(crate) run_id: String,
+    pub(crate) activity_id: String,
+    /// `None` for a completed Activity, which closes with a null result.
+    pub(crate) failure: Option<Failure>,
+}
+
+impl ActivityStreamClose {
+    /// The close for `resolution`, or `None` for one that does not end the Activity.
+    pub(crate) fn of(
+        workflow_id: &str,
+        run_id: &str,
+        activity_id: &str,
+        resolution: Option<&ActivityResolution>,
+    ) -> Option<Self> {
+        let failure = match resolution?.status.as_ref()? {
+            activity_resolution::Status::Completed(_) => None,
+            // A resolution with no failure still ends the Activity, so its streams close with a
+            // failure of their own.
+            activity_resolution::Status::Failed(failed) => Some(
+                failed
+                    .failure
+                    .clone()
+                    .unwrap_or_else(|| no_failure("failed")),
+            ),
+            activity_resolution::Status::Cancelled(cancelled) => Some(
+                cancelled
+                    .failure
+                    .clone()
+                    .unwrap_or_else(|| no_failure("was cancelled")),
+            ),
+            // Only Local Activities back off, and they own no streams.
+            activity_resolution::Status::Backoff(_) => return None,
+        };
+        Some(Self {
+            workflow_id: workflow_id.to_string(),
+            run_id: run_id.to_string(),
+            activity_id: activity_id.to_string(),
+            failure,
+        })
+    }
+}
+
+fn no_failure(what: &str) -> Failure {
+    Failure {
+        message: format!("The Activity {what} with no failure"),
+        ..Default::default()
+    }
+}
+
 /// The records one activation committed, appended when its completion reaches Core.
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct OutputBatch {
@@ -276,6 +339,23 @@ impl OutputStore {
         Ok(dropped)
     }
 
+    /// Closes every topic each Workflow Activity in `closes` wrote, in the background. A close
+    /// that fails is logged and dropped, since it must not fail the Workflow Task. The owner
+    /// checks then close the streams with a generic failure.
+    pub(crate) fn close_activity_streams(&self, closes: Vec<ActivityStreamClose>) {
+        if closes.is_empty() {
+            return;
+        }
+        let store = self.store.clone();
+        let namespace = self.namespace.clone();
+        // Replay never sends the close again, so a Worker shutdown waits for it.
+        self.tasks.spawn(async move {
+            for close in closes {
+                close_one(store.as_ref(), &namespace, close).await;
+            }
+        });
+    }
+
     /// Drops the store's dedupe sessions of attempts that will not write again, in the
     /// background, so the next activation doesn't wait a store round trip per session. A
     /// failure only leaves a session in the store until retention, so it is logged. Then closes
@@ -318,11 +398,115 @@ impl OutputStore {
     }
 }
 
+#[cfg(feature = "streams")]
+async fn close_one(
+    store: &dyn temporalio_streams::StreamStore,
+    namespace: &str,
+    close: ActivityStreamClose,
+) {
+    use temporalio_streams::proto::{ChainId, StoredClose, StreamOwnerKind};
+    let chain = ChainId {
+        namespace: namespace.to_string(),
+        workflow_id: close.workflow_id,
+        first_run_id: close.run_id,
+        owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+        activity_id: close.activity_id,
+    };
+    let closed_by = format!("{}@{}", chain.activity_id, chain.first_run_id);
+    let (result, failure) = match close.failure {
+        None => (Some(temporalio_streams::empty_close_result()), None),
+        Some(failure) => (None, Some(fit_failure(failure, closed_by.len()))),
+    };
+    let topics = match store.written_topics(&chain).await {
+        Ok(topics) => topics,
+        Err(error) => {
+            warn!(
+                workflow_id = %chain.workflow_id,
+                activity_id = %chain.activity_id,
+                %error,
+                "Could not list the topics of an Activity's streams to close them"
+            );
+            return;
+        }
+    };
+    for topic in topics {
+        let stored = StoredClose {
+            result: result.clone(),
+            failure: failure.clone(),
+            closed_by: closed_by.clone(),
+        };
+        if let Err(error) = store.close_stream(&chain, &topic, stored).await {
+            warn!(
+                workflow_id = %chain.workflow_id,
+                activity_id = %chain.activity_id,
+                %topic,
+                %error,
+                "Could not close an Activity's stream when its Workflow resolved it"
+            );
+        }
+    }
+}
+
+/// `failure` cut to what a stored close holds. The causes and the stack trace go first, then the
+/// details, and a failure still too large keeps only the start of its message.
+#[cfg(feature = "streams")]
+fn fit_failure(failure: Failure, closed_by_bytes: usize) -> Failure {
+    use prost::Message;
+    // Room for the close's other fields and their tags.
+    let limit = temporalio_streams::MAX_CLOSE_BYTES - closed_by_bytes - 64;
+    if failure.encoded_len() <= limit {
+        return failure;
+    }
+    let mut cut = Failure {
+        cause: None,
+        stack_trace: String::new(),
+        ..failure
+    };
+    if cut.encoded_len() <= limit {
+        return cut;
+    }
+    cut.encoded_attributes = None;
+    cut.failure_info = None;
+    if cut.encoded_len() <= limit {
+        return cut;
+    }
+    let mut end = limit / 2;
+    while !cut.message.is_char_boundary(end) {
+        end -= 1;
+    }
+    cut.message.truncate(end);
+    Failure {
+        message: cut.message,
+        source: cut.source,
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use temporalio_common::protos::temporal::api::common::v1::Payload;
 
+    #[test]
+    fn a_failed_or_cancelled_resolution_with_no_failure_still_closes() {
+        use temporalio_common::protos::coresdk::activity_result::{
+            Cancellation, Failure as Failed,
+        };
+        for status in [
+            activity_resolution::Status::Failed(Failed {
+                failure: None,
+                ..Default::default()
+            }),
+            activity_resolution::Status::Cancelled(Cancellation { failure: None }),
+        ] {
+            let resolution = ActivityResolution {
+                status: Some(status),
+            };
+            let close = ActivityStreamClose::of("wf", "run-1", "tool", Some(&resolution)).unwrap();
+            assert!(close.failure.is_some());
+        }
+    }
+
     fn data(topic: &str, hash: u8) -> OutputRecord {
         OutputRecord {
             topic: topic.to_string(),
diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md
index 2560d911d5..2c3c629d63 100644
--- a/crates/streams/docs/redis.md
+++ b/crates/streams/docs/redis.md
@@ -76,8 +76,17 @@ the tombstone that tells a log that expired from one that never existed.
 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.
 
+**Workflow Activity owners.** An Activity that a Workflow run scheduled is a chain of one too.
+Its keys have the hash tag `{<namespace>:!wa:<workflow id>:<run id>:<activity id>}`, so each
+Activity lands in a slot of its own and Activity `1` of the next run is another chain. Such a
+chain has one more key, `<prefix>:{<tag>}:topics`, a set of the log keys its producers wrote. The
+append script adds the log key with `SADD` on every batch, since a first batch can be lost, and
+slides the set's expiry to the retention plus the grace on each append. The scheduling Workflow's
+Core reads it with `SMEMBERS` to close every topic when the Activity ends.
+
 **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
+the Workflow's chains or all the Activity's runs. A Workflow's delete also scans for the keys of
+every Activity it scheduled, in all its 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.
 
@@ -91,7 +100,7 @@ percent-encoded. A Redis ACL user for one namespace's applications, with the rul
 conformance suite runs as:
 
 ```text
-ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots
+ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +sadd +smembers +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots
 ```
 
 A cluster client reads the topology with `CLUSTER SLOTS` first. An ACL user lives on one node,
diff --git a/crates/streams/lua/append.lua b/crates/streams/lua/append.lua
index ed7f5b33bf..bf0c82aea9 100644
--- a/crates/streams/lua/append.lua
+++ b/crates/streams/lua/append.lua
@@ -35,6 +35,12 @@ for i = 6, #ARGV do
     first = last
   end
 end
+local topics = KEYS[5]
+if topics then
+  -- On every batch, since the first one can be lost and the close must still find the topic.
+  redis.call('SADD', topics, KEYS[1])
+  redis.call('PEXPIRE', topics, retention + grace)
+end
 keep(KEYS[1], KEYS[2], #ARGV - 5, last, grace, floor_ms, life)
 redis.call('HSET', KEYS[2], field,
   ARGV[4] .. '|' .. (#ARGV - 5) .. '|' .. first .. '|' .. last .. '|' .. digest_arg)
diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs
index 163aa6d20e..21c31d2e9f 100644
--- a/crates/streams/src/driver.rs
+++ b/crates/streams/src/driver.rs
@@ -437,7 +437,7 @@ impl StreamDriver {
             .map(|((chain, topic), (counter, position, ids))| {
                 let chain = chain.into_chain();
                 let (kind, id) = chain_owner(&chain);
-                let hash = stream_hash(&chain.namespace, kind, id, &topic);
+                let hash = stream_hash(&chain.namespace, kind, &id, &topic);
                 let notification = StreamNotification {
                     position: mint_cursor(self.store.name(), &hash, &position),
                     chain,
diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs
index cf8252ff3f..bab8a5b038 100644
--- a/crates/streams/src/handle.rs
+++ b/crates/streams/src/handle.rs
@@ -79,32 +79,48 @@ impl StreamsOptions {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+enum Kind {
+    Workflow,
+    Activity,
+    WorkflowActivity,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct OwnerKey {
     namespace: String,
-    activity: bool,
-    /// The Workflow id or the Activity id.
+    kind: Kind,
+    /// The Workflow id, the standalone Activity id, or the Workflow that scheduled a Workflow
+    /// Activity.
     id: String,
+    /// A Workflow Activity's id. Empty for other kinds.
+    activity_id: String,
+    /// The pinned run. A Workflow Activity always names its scheduling run.
     run_id: String,
 }
 
 impl OwnerKey {
     fn owner(&self) -> OwnerId<'_> {
-        if self.activity {
-            OwnerId::Activity(&self.id)
-        } else {
-            OwnerId::Workflow(&self.id)
+        match self.kind {
+            Kind::Workflow => OwnerId::Workflow(&self.id),
+            Kind::Activity => OwnerId::Activity(&self.id),
+            Kind::WorkflowActivity => OwnerId::WorkflowActivity {
+                workflow_id: &self.id,
+                activity_id: &self.activity_id,
+            },
         }
     }
 
     /// The owner as error texts name it.
     fn label(&self) -> String {
-        let kind = if self.activity {
-            "Activity"
-        } else {
-            "Workflow"
-        };
-        format!("{kind} {:?}", self.id)
+        match self.kind {
+            Kind::Workflow => format!("Workflow {:?}", self.id),
+            Kind::Activity => format!("Activity {:?}", self.id),
+            Kind::WorkflowActivity => format!(
+                "Activity {:?} of run {:?} of Workflow {:?}",
+                self.activity_id, self.run_id, self.id
+            ),
+        }
     }
 }
 
@@ -387,7 +403,7 @@ impl Streams {
                 break page;
             }
             if unix_ms() >= state.next_check_ms {
-                state.owner_ended = self.owner_ended(stream, &target.chain).await?;
+                state.owner_ended = self.owner_ended(stream, &target.chain, &mut state).await?;
                 state.check_interval_ms = (state.check_interval_ms * 2)
                     .clamp(millis(minimum), millis(self.options.owner_check_max));
                 state.next_check_ms = unix_ms() + state.check_interval_ms;
@@ -617,11 +633,35 @@ impl Streams {
     }
 
     /// Whether the owner a read follows ended. Following the chain, a run that continued as
-    /// new is not the end, and a new chain on the Workflow id is. A reader only ends its own read,
-    /// so it needs no write access. Producers and the owner's Core close the chain.
-    async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult<bool> {
+    /// new is not the end, and a new chain on the Workflow id is. A Workflow Activity that its
+    /// running run no longer lists as pending ends the read only after
+    /// [StreamsOptions::owner_start_grace] from the first check that found it so, since a
+    /// Workflow can write a link to the stream before the server takes the Activity's schedule.
+    /// A reader only ends its own read, so it needs no write access. Producers and the owner's
+    /// Core close the chain.
+    async fn owner_ended(
+        &self,
+        stream: &StreamAddress,
+        chain: &ChainId,
+        state: &mut ReadState,
+    ) -> StreamResult<bool> {
         let key = owner_key(stream);
         match self.describe(&key, &key.run_id).await {
+            Ok(described) if key.kind == Kind::WorkflowActivity => match described.status {
+                OwnerStatus::Running => {
+                    state.owner_missing_since_ms = 0;
+                    Ok(false)
+                }
+                OwnerStatus::ActivityGone => {
+                    let now = unix_ms();
+                    if state.owner_missing_since_ms == 0 {
+                        state.owner_missing_since_ms = now;
+                    }
+                    let gone_for = now.saturating_sub(state.owner_missing_since_ms);
+                    Ok(gone_for >= millis(self.options.owner_start_grace))
+                }
+                OwnerStatus::ContinuedAsNew | OwnerStatus::Ended => Ok(true),
+            },
             // 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)),
@@ -743,6 +783,11 @@ impl Streams {
         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),
+            Ok(StreamOwnerKind::WorkflowActivity) => {
+                return Err(StreamError::invalid_argument(
+                    "a Workflow Activity's streams go with its Workflow's delete",
+                ));
+            }
             _ => return Err(unsupported_kind(request.owner_kind)),
         };
         if request.namespace.is_empty() || id.is_empty() {
@@ -783,7 +828,21 @@ impl Streams {
         {
             return Ok(chain_of_key(owner, &checked.first_run_id));
         }
-        let chain = if owner.run_id.is_empty() {
+        let chain = if owner.kind == Kind::WorkflowActivity {
+            // A Workflow Activity writes only while it runs, so a run that no longer lists it as
+            // pending refuses at once.
+            let chain = chain_of_key(owner, &owner.run_id);
+            match self.describe(owner, &owner.run_id).await {
+                Err(OwnerError::NotFound(_)) if known.is_some() => {
+                    return Err(self.closed(&chain, gone_text(owner), writer).await);
+                }
+                Err(error) => return Err(not_found(owner, error)),
+                Ok(described) if described.status != OwnerStatus::Running => {
+                    return Err(self.closed(&chain, "has ended", writer).await);
+                }
+                Ok(_) => chain,
+            }
+        } else if owner.run_id.is_empty() {
             // 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) {
@@ -936,7 +995,7 @@ impl Streams {
     fn hash(&self, stream: &StreamAddress) -> String {
         let chain = chain_of(stream, "");
         let (kind, id) = chain_owner(&chain);
-        stream_hash(&stream.namespace, kind, id, &stream.topic)
+        stream_hash(&stream.namespace, kind, &id, &stream.topic)
     }
 
     fn cursor(&self, stream: &StreamAddress, position: &str) -> String {
@@ -995,6 +1054,24 @@ pub(crate) fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAdd
                 ));
             }
         }
+        Ok(StreamOwnerKind::WorkflowActivity) => {
+            if stream.namespace.is_empty()
+                || stream.workflow_id.is_empty()
+                || stream.activity_id.is_empty()
+            {
+                return Err(StreamError::invalid_argument(
+                    "a stream that a Workflow Activity owns needs its namespace, its Workflow id \
+                     and its Activity id",
+                ));
+            }
+            if stream.run_id.is_empty() {
+                // An Activity id names nothing across runs, since ids such as `1` repeat after
+                // Continue-as-New.
+                return Err(StreamError::invalid_argument(
+                    "a stream that a Workflow Activity owns needs the run that scheduled it",
+                ));
+            }
+        }
         _ => return Err(unsupported_kind(stream.owner_kind)),
     }
     check_topic(&stream.topic)?;
@@ -1003,8 +1080,8 @@ pub(crate) fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAdd
 
 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}"
+        "this release keeps streams of Workflows, standalone Activities and Workflow Activities \
+         only, not owner kind {owner_kind}"
     ))
 }
 
@@ -1078,28 +1155,48 @@ fn millis(duration: Duration) -> u64 {
     u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
 }
 
+fn kind_of(owner_kind: i32) -> Kind {
+    match StreamOwnerKind::try_from(owner_kind) {
+        Ok(StreamOwnerKind::Activity) => Kind::Activity,
+        Ok(StreamOwnerKind::WorkflowActivity) => Kind::WorkflowActivity,
+        _ => Kind::Workflow,
+    }
+}
+
 fn owner_key(stream: &StreamAddress) -> OwnerKey {
-    let activity = stream.owner_kind == StreamOwnerKind::Activity as i32;
+    let kind = kind_of(stream.owner_kind);
+    let (id, activity_id) = match kind {
+        Kind::Workflow => (&stream.workflow_id, ""),
+        Kind::Activity => (&stream.activity_id, ""),
+        Kind::WorkflowActivity => (&stream.workflow_id, stream.activity_id.as_str()),
+    };
     OwnerKey {
         namespace: stream.namespace.clone(),
-        activity,
-        id: if activity {
-            stream.activity_id.clone()
-        } else {
-            stream.workflow_id.clone()
-        },
+        kind,
+        id: id.clone(),
+        activity_id: activity_id.to_string(),
         run_id: stream.run_id.clone(),
     }
 }
 
-/// The latest run of the owner of `chain`.
+/// The latest run of the owner of `chain`, or a Workflow Activity's scheduling run.
 fn chain_owner_key(chain: &ChainId) -> OwnerKey {
-    let activity = chain.owner_kind == StreamOwnerKind::Activity as i32;
+    let kind = kind_of(chain.owner_kind);
+    let (id, activity_id, run_id) = match kind {
+        Kind::Workflow => (&chain.workflow_id, "", ""),
+        Kind::Activity => (&chain.activity_id, "", ""),
+        Kind::WorkflowActivity => (
+            &chain.workflow_id,
+            chain.activity_id.as_str(),
+            chain.first_run_id.as_str(),
+        ),
+    };
     OwnerKey {
         namespace: chain.namespace.clone(),
-        activity,
-        id: chain_owner(chain).1.to_string(),
-        run_id: String::new(),
+        kind,
+        id: id.clone(),
+        activity_id: activity_id.to_string(),
+        run_id: run_id.to_string(),
     }
 }
 
@@ -1111,34 +1208,43 @@ pub(crate) fn chain_of(stream: &StreamAddress, first_run_id: &str) -> ChainId {
 fn chain_of_key(owner: &OwnerKey, first_run_id: &str) -> ChainId {
     // 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 {
+    match owner.kind {
+        Kind::Workflow => ChainId {
+            namespace: owner.namespace.clone(),
+            workflow_id: owner.id.clone(),
+            first_run_id: first_run_id.to_string(),
+            ..Default::default()
+        },
+        Kind::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 {
+        },
+        // The scheduling run keys a Workflow Activity, whatever chain that run is in.
+        Kind::WorkflowActivity => ChainId {
             namespace: owner.namespace.clone(),
             workflow_id: owner.id.clone(),
-            first_run_id: first_run_id.to_string(),
-            ..Default::default()
-        }
+            first_run_id: owner.run_id.clone(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            activity_id: owner.activity_id.clone(),
+        },
     }
 }
 
 /// 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"
+    match owner.kind {
+        Kind::Activity => "is gone from Temporal",
+        Kind::Workflow | Kind::WorkflowActivity => "is gone from History",
     }
 }
 
 fn first_run_id(owner: &OwnerKey, described: &OwnerDescription) -> StreamResult<String> {
+    if owner.kind == Kind::WorkflowActivity {
+        return Ok(owner.run_id.clone());
+    }
     if described.first_run_id.is_empty() {
         // 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.
@@ -1199,6 +1305,8 @@ mod tests {
     struct Owners {
         latest: Mutex<HashMap<String, Result<OwnerDescription, OwnerError>>>,
         runs: Mutex<HashMap<String, OwnerDescription>>,
+        /// The Activities each run lists as pending, by run id and Activity id.
+        pending: Mutex<std::collections::HashSet<(String, String)>>,
         describes: AtomicUsize,
     }
 
@@ -1255,6 +1363,17 @@ mod tests {
         fn describes(&self) -> usize {
             self.describes.load(Ordering::Relaxed)
         }
+
+        /// Sets whether run `run_id` lists Activity `activity_id` as pending.
+        fn pending(&self, run_id: &str, activity_id: &str, pending: bool) {
+            let key = (run_id.to_string(), activity_id.to_string());
+            let mut all = self.pending.lock().unwrap();
+            if pending {
+                all.insert(key);
+            } else {
+                all.remove(&key);
+            }
+        }
     }
 
     #[async_trait::async_trait]
@@ -1269,6 +1388,26 @@ mod tests {
             let workflow_id = &match owner {
                 OwnerId::Workflow(id) => id.to_string(),
                 OwnerId::Activity(id) => format!("activity/{id}"),
+                OwnerId::WorkflowActivity { activity_id, .. } => {
+                    let run = self
+                        .runs
+                        .lock()
+                        .unwrap()
+                        .get(run_id)
+                        .cloned()
+                        .ok_or_else(|| OwnerError::NotFound(run_id.to_string()))?;
+                    let pending = self
+                        .pending
+                        .lock()
+                        .unwrap()
+                        .contains(&(run_id.to_string(), activity_id.to_string()));
+                    let status = match run.status {
+                        OwnerStatus::Running if pending => OwnerStatus::Running,
+                        OwnerStatus::Running => OwnerStatus::ActivityGone,
+                        _ => OwnerStatus::Ended,
+                    };
+                    return Ok(OwnerDescription { status, ..run });
+                }
             };
             if !run_id.is_empty() {
                 return self
@@ -3320,4 +3459,332 @@ mod tests {
         let (records, _) = shape(&setup, &page.cursor, &page.state).await;
         assert_eq!(records, [("a2".to_string(), key(1, 0), 0, true)]);
     }
+
+    /// A stream of Activity `tool` that run `run-1` of Workflow `wf` scheduled.
+    fn wfa_stream(topic: &str) -> StreamAddress {
+        StreamAddress {
+            namespace: "ns".to_string(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            workflow_id: "wf".to_string(),
+            run_id: "run-1".to_string(),
+            activity_id: "tool".to_string(),
+            topic: topic.to_string(),
+        }
+    }
+
+    fn wfa_chain(run_id: &str) -> ChainId {
+        ChainId {
+            namespace: "ns".to_string(),
+            workflow_id: "wf".to_string(),
+            first_run_id: run_id.to_string(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            activity_id: "tool".to_string(),
+        }
+    }
+
+    /// Run `run-1` of `wf` lists `tool` as pending.
+    fn wfa_setup(options: StreamsOptions) -> Setup {
+        let setup = setup(options);
+        setup.owners.pending("run-1", "tool", true);
+        setup
+    }
+
+    fn wfa_append(sequence: i64, value: &str) -> AppendRequest {
+        AppendRequest {
+            stream: Some(wfa_stream("out")),
+            producer: Some(Producer::Activity(ActivityProducer {
+                workflow_id: "wf".to_string(),
+                run_id: "run-1".to_string(),
+                activity_id: "tool".to_string(),
+                attempt: 1,
+                ..Default::default()
+            })),
+            ..append("", sequence, value)
+        }
+    }
+
+    fn wfa_read(after: &str, state: &[u8], wait: Duration) -> ReadRequest {
+        ReadRequest {
+            stream: Some(wfa_stream("out")),
+            ..read_after(after, state, wait)
+        }
+    }
+
+    fn end_run(setup: &Setup, run_id: &str) {
+        setup
+            .owners
+            .runs
+            .lock()
+            .unwrap()
+            .get_mut(run_id)
+            .unwrap()
+            .status = OwnerStatus::Ended;
+    }
+
+    async fn wfa_chain_closed(store: &MemoryStore, run_id: &str) -> bool {
+        store
+            .read(StoreReadRequest {
+                chain: Some(wfa_chain(run_id)),
+                topic: "out".to_string(),
+                max_records: 1,
+                ..Default::default()
+            })
+            .await
+            .unwrap()
+            .closed
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_writes_its_own_stream_keyed_by_its_run() {
+        let setup = wfa_setup(quick());
+        let landed = setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        let held = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(wfa_chain("run-1")),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(held.position, "0");
+        assert_eq!(
+            setup
+                .store
+                .written_topics(&wfa_chain("run-1"))
+                .await
+                .unwrap(),
+            ["out"]
+        );
+        let read = setup
+            .streams
+            .read(wfa_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, "run-1");
+        // Activity ids repeat after Continue-as-New, so the next run's `tool` is another stream.
+        setup.owners.run("wf", "run-2", "run-1");
+        setup.owners.pending("run-2", "tool", true);
+        let mut next = wfa_stream("out");
+        next.run_id = "run-2".to_string();
+        let error = setup
+            .streams
+            .read(ReadRequest {
+                stream: Some(next),
+                after: landed.last_cursor.clone(),
+                ..Default::default()
+            })
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Cursor, "{error}");
+        // The Workflow's own stream on the same topic is another stream too.
+        let error = setup
+            .streams
+            .read(ReadRequest {
+                stream: Some(stream("out")),
+                after: landed.last_cursor,
+                ..Default::default()
+            })
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Cursor, "{error}");
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_address_needs_its_workflow_run_and_activity() {
+        let setup = wfa_setup(quick());
+        let mut bad = Vec::new();
+        for clear in [
+            |s: &mut StreamAddress| s.workflow_id.clear(),
+            |s: &mut StreamAddress| s.run_id.clear(),
+            |s: &mut StreamAddress| s.activity_id.clear(),
+            |s: &mut StreamAddress| s.namespace.clear(),
+        ] {
+            let mut stream = wfa_stream("out");
+            clear(&mut stream);
+            bad.push(stream);
+        }
+        for bad in bad {
+            assert_eq!(
+                kind(
+                    setup
+                        .streams
+                        .latest(LatestRequest { stream: Some(bad) })
+                        .await
+                ),
+                StreamFailureKind::InvalidArgument
+            );
+        }
+        assert_eq!(setup.owners.describes(), 0);
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_read_ends_done_at_a_stored_close() {
+        let setup = wfa_setup(quick());
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        setup
+            .streams
+            .close(CloseRequest {
+                stream: Some(wfa_stream("out")),
+                closed_by: "tool@run-1".to_string(),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        let (delivered, end) =
+            read_activity_to_end(&setup.streams, wfa_read("", &[], Duration::from_secs(5))).await;
+        assert_eq!(delivered, 1);
+        assert!(end.done);
+        assert_eq!(end.closed_by, "tool@run-1");
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_read_ends_at_once_when_its_run_closed() {
+        let setup = wfa_setup(quick());
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        end_run(&setup, "run-1");
+        let started = std::time::Instant::now();
+        let (delivered, end) =
+            read_activity_to_end(&setup.streams, wfa_read("", &[], Duration::from_secs(5))).await;
+        assert_eq!(delivered, 1);
+        assert!(end.done);
+        assert!(started.elapsed() < Duration::from_secs(2));
+        // A pinned run's end marks nothing closed, as for other pinned reads.
+        assert!(!wfa_chain_closed(&setup.store, "run-1").await);
+    }
+
+    #[tokio::test]
+    async fn a_gone_workflow_activity_ends_a_read_only_after_the_grace() {
+        let grace = Duration::from_millis(400);
+        let setup = wfa_setup(StreamsOptions {
+            owner_start_grace: grace,
+            ..quick()
+        });
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        setup.owners.pending("run-1", "tool", false);
+        let first = setup
+            .streams
+            .read(wfa_read("", &[], Duration::ZERO))
+            .await
+            .unwrap();
+        let started = std::time::Instant::now();
+        let idle = setup
+            .streams
+            .read(wfa_read(
+                &first.cursor,
+                &first.state,
+                Duration::from_millis(150),
+            ))
+            .await
+            .unwrap();
+        assert!(!idle.done);
+        let (delivered, end) = read_activity_to_end(
+            &setup.streams,
+            wfa_read(&idle.cursor, &idle.state, Duration::from_secs(5)),
+        )
+        .await;
+        assert_eq!(delivered, 0);
+        assert!(end.done);
+        assert!(started.elapsed() >= grace - Duration::from_millis(100));
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_read_before_its_schedule_stays_open() {
+        let setup = setup(quick());
+        let idle = setup
+            .streams
+            .read(wfa_read("", &[], Duration::from_millis(150)))
+            .await
+            .unwrap();
+        assert!(idle.records.is_empty() && !idle.done);
+        setup.owners.pending("run-1", "tool", true);
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        let read = setup
+            .streams
+            .read(wfa_read(&idle.cursor, &idle.state, Duration::ZERO))
+            .await
+            .unwrap();
+        assert_eq!(read.records.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_read_of_a_missing_run_waits_the_grace() {
+        let setup = before_owner_starts(Duration::ZERO);
+        let error = setup
+            .streams
+            .read(wfa_read("", &[], Duration::from_secs(5)))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::NotFound);
+        assert!(error.message.contains("Activity \"tool\""), "{error}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_workflow_activity_producer_is_refused_once_the_activity_is_gone() {
+        let setup = wfa_setup(quick());
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        setup.owners.pending("run-1", "tool", false);
+        tokio::time::advance(OWNER_RECHECK + Duration::from_secs(1)).await;
+        let error = setup.streams.append(wfa_append(2, "2")).await.unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::Closed);
+        assert!(error.message.contains("Activity \"tool\""), "{error}");
+        assert!(wfa_chain_closed(&setup.store, "run-1").await);
+    }
+
+    #[tokio::test]
+    async fn a_workflow_activity_producer_of_a_closed_run_is_refused() {
+        let setup = wfa_setup(quick());
+        end_run(&setup, "run-1");
+        assert_eq!(
+            kind(setup.streams.append(wfa_append(1, "1")).await),
+            StreamFailureKind::Closed
+        );
+    }
+
+    #[tokio::test]
+    async fn a_workflow_delete_removes_the_streams_of_its_workflow_activities() {
+        let setup = wfa_setup(quick());
+        setup.streams.append(wfa_append(1, "1")).await.unwrap();
+        let mut request = DeleteOwnerRequest {
+            namespace: "ns".to_string(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            workflow_id: "wf".to_string(),
+            ..Default::default()
+        };
+        assert_eq!(
+            kind(setup.streams.delete_owner(request.clone()).await),
+            StreamFailureKind::InvalidArgument
+        );
+        request.owner_kind = StreamOwnerKind::Workflow as i32;
+        // A Workflow whose id only starts with `wf` keeps its Activities' streams.
+        setup.owners.run("wf2", "run-9", "run-9");
+        setup.owners.pending("run-9", "tool", true);
+        let mut other = wfa_append(1, "1");
+        other.stream.as_mut().unwrap().workflow_id = "wf2".to_string();
+        other.stream.as_mut().unwrap().run_id = "run-9".to_string();
+        setup.streams.append(other).await.unwrap();
+        assert!(setup.streams.delete_owner(request).await.unwrap().deleted > 0);
+        let gone = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(wfa_chain("run-1")),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(gone.position, "");
+        let kept = setup
+            .store
+            .latest(StoreLatestRequest {
+                chain: Some(ChainId {
+                    workflow_id: "wf2".to_string(),
+                    ..wfa_chain("run-9")
+                }),
+                topic: "out".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(kept.position, "0");
+    }
 }
diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs
index 69f6956e0a..bf84abb16f 100644
--- a/crates/streams/src/lib.rs
+++ b/crates/streams/src/lib.rs
@@ -45,9 +45,10 @@ pub use reader::{
     DEFAULT_MAX_RECORDS, MAX_STATE_BYTES, MAX_TRACKED_PRODUCERS, ReadTarget, read_page,
 };
 pub use record::{
-    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,
+    ACTIVITY_OWNER_KIND, MAX_KEYED_ACTIVITY_ATTEMPT, WORKFLOW_ACTIVITY_OWNER_KIND,
+    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 4ecc266e78..7996c0f0d9 100644
--- a/crates/streams/src/memory.rs
+++ b/crates/streams/src/memory.rs
@@ -10,15 +10,16 @@
 use crate::{
     ACTIVITY_OWNER_KIND, Capabilities, ConditionForms, MAX_BATCH_BYTES, MAX_RECORD_BYTES,
     NotificationEntry, NotificationLog, StreamError, StreamResult, StreamStore,
-    WORKFLOW_OWNER_KIND, chain_owner,
+    WORKFLOW_ACTIVITY_OWNER_KIND, WORKFLOW_OWNER_KIND, chain_owner,
     proto::{
         ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse,
         StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose,
         StoredRecord, StreamOwnerKind,
     },
+    record::part,
 };
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{BTreeSet, HashMap, VecDeque},
     sync::{Arc, Mutex},
     time::Duration,
 };
@@ -104,6 +105,8 @@ struct Chain {
     /// Per topic, the position each positions-only reader holds the records after. `None` holds
     /// the whole topic.
     floors: HashMap<String, HashMap<String, Option<u64>>>,
+    /// The topics producers wrote, for a Workflow Activity chain only.
+    written: BTreeSet<String>,
 }
 
 /// A [StreamStore] in process memory. For tests only.
@@ -413,11 +416,11 @@ impl StreamStore for MemoryStore {
         if request.digest.is_empty() {
             return Err(StreamError::refused("an append needs the batch's digest"));
         }
+        let id = chain_of(request.chain.as_ref())?;
+        let keeps_topics = id.owner_kind == StreamOwnerKind::WorkflowActivity as i32;
         let producer_state = self.capabilities.producer_state;
         let mut chains = self.chains.lock().unwrap();
-        let chain = chains
-            .entry(chain_key(chain_of(request.chain.as_ref())?))
-            .or_default();
+        let chain = chains.entry(chain_key(id)).or_default();
         let chain_closed = chain.closed;
         let session = (request.producer_id.clone(), request.attempt);
         let existing = chain.topics.get(&request.topic).filter(|_| producer_state);
@@ -460,6 +463,10 @@ impl StreamStore for MemoryStore {
                     .with_closed_by(closed_by),
             );
         }
+        // On every batch, since the first one can be lost and the close must still find it.
+        if keeps_topics {
+            chain.written.insert(request.topic.clone());
+        }
         let topic = chain.topics.entry(request.topic.clone()).or_default();
         let count = request.records.len() as i64;
         let mut positions = request.records.into_iter().map(|record| topic.push(record));
@@ -670,6 +677,14 @@ impl StreamStore for MemoryStore {
         Ok(())
     }
 
+    async fn written_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
+        let chains = self.chains.lock().unwrap();
+        Ok(chains
+            .get(&chain_key(chain))
+            .map(|chain| chain.written.iter().cloned().collect())
+            .unwrap_or_default())
+    }
+
     async fn list_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
         if !self.capabilities.topic_listing {
             return Err(StreamError::unsupported(
@@ -707,10 +722,17 @@ impl StreamStore for MemoryStore {
                 )));
             }
         };
+        // A Workflow Activity's id starts with its Workflow's, so a Workflow's delete takes it too.
+        let scheduled = (owner_kind == WORKFLOW_OWNER_KIND).then(|| format!("{}/", part(owner_id)));
         let mut chains = self.chains.lock().unwrap();
         let mut deleted = 0;
         chains.retain(|(namespace, kind, id, _), chain| {
-            let owned = *namespace == request.namespace && *kind == owner_kind && id == owner_id;
+            let owned = *namespace == request.namespace
+                && ((*kind == owner_kind && id == owner_id)
+                    || (*kind == WORKFLOW_ACTIVITY_OWNER_KIND
+                        && scheduled
+                            .as_ref()
+                            .is_some_and(|prefix| id.starts_with(prefix))));
             if owned {
                 deleted += 1 + chain.topics.len() as u64;
             }
diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs
index 5b0cfd0ea7..06a851050e 100644
--- a/crates/streams/src/notify.rs
+++ b/crates/streams/src/notify.rs
@@ -697,7 +697,7 @@ fn cursor_of(store: &dyn StreamStore, chain: &ChainId, topic: &str, position: &s
     let (kind, id) = chain_owner(chain);
     mint_cursor(
         store.name(),
-        &stream_hash(&chain.namespace, kind, id, topic),
+        &stream_hash(&chain.namespace, kind, &id, topic),
         position,
     )
 }
@@ -812,6 +812,10 @@ impl StreamStore for NotifyingStore {
             .await
     }
 
+    async fn written_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
+        self.inner.written_topics(chain).await
+    }
+
     async fn list_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
         self.inner.list_topics(chain).await
     }
@@ -1291,6 +1295,9 @@ mod tests {
         async fn retire_producer(&self, c: &ChainId, t: &str, p: &str, a: i64) -> StreamResult<()> {
             self.0.retire_producer(c, t, p, a).await
         }
+        async fn written_topics(&self, c: &ChainId) -> StreamResult<Vec<String>> {
+            self.0.written_topics(c).await
+        }
         async fn delete_owner(&self, r: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
             self.0.delete_owner(r).await
         }
diff --git a/crates/streams/src/owner.rs b/crates/streams/src/owner.rs
index 14ff04eaf5..9692e0a92b 100644
--- a/crates/streams/src/owner.rs
+++ b/crates/streams/src/owner.rs
@@ -15,6 +15,13 @@ pub enum OwnerId<'a> {
     Workflow(&'a str),
     /// A standalone Activity id.
     Activity(&'a str),
+    /// An Activity that a Workflow run scheduled. The run is the one a describe names.
+    WorkflowActivity {
+        /// The scheduling Workflow.
+        workflow_id: &'a str,
+        /// The Activity id in the scheduling run.
+        activity_id: &'a str,
+    },
 }
 
 /// How an owner's run stands, as the stream layer needs it.
@@ -26,6 +33,9 @@ pub enum OwnerStatus {
     ContinuedAsNew,
     /// The run closed and ended its chain.
     Ended,
+    /// The scheduling run of a Workflow Activity runs, but the Activity is not pending in it. It
+    /// either ended or is not scheduled yet, and the describe can't tell which.
+    ActivityGone,
 }
 
 impl OwnerStatus {
@@ -48,6 +58,17 @@ impl OwnerStatus {
             Unspecified | Running | Paused => Self::Running,
         }
     }
+
+    /// The status of a Workflow Activity, from its scheduling run's status and whether the run
+    /// lists the Activity as pending. A closed run has no pending Activities, so it ends the
+    /// Activity's streams, also when it continued as new.
+    pub fn of_workflow_activity(status: WorkflowExecutionStatus, pending: bool) -> Self {
+        match Self::of_workflow(status) {
+            Self::Running if pending => Self::Running,
+            Self::Running => Self::ActivityGone,
+            _ => Self::Ended,
+        }
+    }
 }
 
 /// One run of an owner, as Temporal describes it.
@@ -56,7 +77,8 @@ 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. An Activity run is
-    /// its own chain.
+    /// its own chain. The stream layer keys a Workflow Activity by its scheduling run, whatever
+    /// this holds.
     pub first_run_id: String,
     /// How the run stands.
     pub status: OwnerStatus,
diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs
index 346a022d28..a9b56662ec 100644
--- a/crates/streams/src/reader.rs
+++ b/crates/streams/src/reader.rs
@@ -755,6 +755,10 @@ mod tests {
             unreachable!()
         }
 
+        async fn written_topics(&self, _: &ChainId) -> StreamResult<Vec<String>> {
+            unreachable!()
+        }
+
         async fn delete_owner(&self, _: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
             unreachable!()
         }
diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs
index 16a13be33c..9270f2f68d 100644
--- a/crates/streams/src/record.rs
+++ b/crates/streams/src/record.rs
@@ -6,6 +6,7 @@ use crate::{
         ActivityProducer, AppendRecord, ChainId, StreamOwnerKind, StreamRecord, StreamRecordKind,
     },
 };
+use std::borrow::Cow;
 use temporalio_common::{
     protos::temporal::api::common::v1::Payload,
     streams::{CONTENT_HASH_KEY, METADATA_ENCODING, content_hash_text},
@@ -17,13 +18,40 @@ 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 Workflow Activity's streams hash under in their cursors.
+pub const WORKFLOW_ACTIVITY_OWNER_KIND: &str = "workflow_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)
+///
+/// A Workflow Activity's id holds the scheduling run, so a cursor of Activity `"1"` in one run is
+/// refused on Activity `"1"` of the next run.
+pub fn chain_owner(chain: &ChainId) -> (&'static str, Cow<'_, str>) {
+    match StreamOwnerKind::try_from(chain.owner_kind) {
+        Ok(StreamOwnerKind::Activity) => (ACTIVITY_OWNER_KIND, Cow::Borrowed(&chain.activity_id)),
+        Ok(StreamOwnerKind::WorkflowActivity) => (
+            WORKFLOW_ACTIVITY_OWNER_KIND,
+            Cow::Owned(format!(
+                "{}/{}/{}",
+                part(&chain.workflow_id),
+                part(&chain.first_run_id),
+                part(&chain.activity_id)
+            )),
+        ),
+        _ => (WORKFLOW_OWNER_KIND, Cow::Borrowed(&chain.workflow_id)),
+    }
+}
+
+/// Percent-encodes every byte but the unreserved ones, as Python's `quote(text, safe="")` does.
+pub(crate) fn part(text: &str) -> String {
+    let mut out = String::with_capacity(text.len());
+    for byte in text.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.' | b'-' | b'~') {
+            out.push(byte as char);
+        } else {
+            out.push_str(&format!("%{byte:02X}"));
+        }
     }
+    out
 }
 
 const HASH_LENGTH: usize = 32;
diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs
index 41f50a8fcb..31a62828ae 100644
--- a/crates/streams/src/redis/keys.rs
+++ b/crates/streams/src/redis/keys.rs
@@ -15,6 +15,15 @@
 //! namespace, `{<namespace>:!a:<activity id>:<run id>}`, and the same keys follow it. Each part
 //! escapes `!`, so no Workflow id makes that tag.
 //!
+//! A Workflow Activity is a chain of one too. Its tag is
+//! `{<namespace>:!wa:<workflow id>:<scheduling run id>:<activity id>}`, so each tool call lands in
+//! a slot of its own. It has one more key, the set of log keys its producers wrote, which the
+//! scheduling Workflow's Core reads to close every topic when the Activity ends:
+//!
+//! ```text
+//! <prefix>:{<namespace>:!wa:<workflow id>:<run id>:<activity id>}:topics
+//! ```
+//!
 //! 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:
@@ -30,23 +39,14 @@
 
 use super::{conn::slot, slots};
 use crate::proto::{ChainId, StreamOwnerKind};
+pub(crate) use crate::record::part;
 
 /// 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 {
-    let mut out = String::with_capacity(text.len());
-    for byte in text.bytes() {
-        if byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.' | b'-' | b'~') {
-            out.push(byte as char);
-        } else {
-            out.push_str(&format!("%{byte:02X}"));
-        }
-    }
-    out
-}
+/// The part after the namespace that marks a Workflow Activity chain's tag.
+const WORKFLOW_ACTIVITY_MARK: &str = "!wa";
 
 /// The keys of one run chain's streams.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -56,21 +56,33 @@ 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)
+        let run = part(&chain.first_run_id);
+        let owner = match StreamOwnerKind::try_from(chain.owner_kind) {
+            Ok(StreamOwnerKind::Activity) => {
+                format!("{ACTIVITY_MARK}:{}:{run}", part(&chain.activity_id))
+            }
+            Ok(StreamOwnerKind::WorkflowActivity) => format!(
+                "{WORKFLOW_ACTIVITY_MARK}:{}:{run}:{}",
+                part(&chain.workflow_id),
+                part(&chain.activity_id)
+            ),
+            _ => format!("{}:{run}", part(&chain.workflow_id)),
         };
         Self {
-            base: format!(
-                "{}:{{{}:{owner}:{}}}",
-                part(prefix),
-                part(&chain.namespace),
-                part(&chain.first_run_id)
-            ),
+            base: format!("{}:{{{}:{owner}}}", part(prefix), part(&chain.namespace)),
         }
     }
 
+    /// Whether the chain is a Workflow Activity's, which keeps the set of topics it wrote.
+    pub(crate) fn keeps_topics(chain: &ChainId) -> bool {
+        chain.owner_kind == StreamOwnerKind::WorkflowActivity as i32
+    }
+
+    /// The log keys a Workflow Activity chain's producers wrote.
+    pub(crate) fn topics(&self) -> String {
+        format!("{}:topics", self.base)
+    }
+
     pub(crate) fn log(&self, topic: &str) -> String {
         format!("{}:t:{}", self.base, part(topic))
     }
@@ -141,6 +153,19 @@ pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)>
             activity_id: unpart(activity_id)?,
             ..Default::default()
         },
+        [
+            namespace,
+            WORKFLOW_ACTIVITY_MARK,
+            workflow_id,
+            run_id,
+            activity_id,
+        ] => ChainId {
+            namespace: unpart(namespace)?,
+            workflow_id: unpart(workflow_id)?,
+            first_run_id: unpart(run_id)?,
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            activity_id: unpart(activity_id)?,
+        },
         _ => return None,
     };
     Some((chain, unpart(topic)?))
@@ -179,6 +204,21 @@ pub(crate) fn owner_pattern(prefix: &str, namespace: &str, workflow_id: &str) ->
     )
 }
 
+/// The pattern that matches every key of the Workflow Activities that one Workflow id scheduled,
+/// across all its runs.
+pub(crate) fn workflow_activity_pattern(
+    prefix: &str,
+    namespace: &str,
+    workflow_id: &str,
+) -> String {
+    format!(
+        "{}:{{{}:{WORKFLOW_ACTIVITY_MARK}:{}:*",
+        part(prefix),
+        part(namespace),
+        part(workflow_id)
+    )
+}
+
 /// 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!(
@@ -216,7 +256,15 @@ mod tests {
     }
 
     fn case_chain(case: &Value) -> ChainId {
-        if case["owner_kind"] == "activity" {
+        if case["owner_kind"] == "workflow_activity" {
+            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(),
+                owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+                activity_id: text(case, "activity_id").to_string(),
+            }
+        } else if case["owner_kind"] == "activity" {
             ChainId {
                 namespace: text(case, "namespace").to_string(),
                 first_run_id: text(case, "first_run_id").to_string(),
@@ -241,12 +289,25 @@ mod tests {
             serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap();
         assert!(cases.len() >= 4);
         assert!(cases.iter().any(|case| case["owner_kind"] == "activity"));
+        assert!(
+            cases
+                .iter()
+                .any(|case| case["owner_kind"] == "workflow_activity")
+        );
         for case in &cases {
-            let keys = ChainKeys::new(text(case, "prefix"), &case_chain(case));
+            let chain = case_chain(case);
+            let keys = ChainKeys::new(text(case, "prefix"), &chain);
             let topic = text(case, "topic");
             assert_eq!(keys.log(topic), text(case, "log"));
             assert_eq!(keys.meta(topic), text(case, "meta"));
             assert_eq!(keys.chain(), text(case, "chain"));
+            assert_eq!(
+                ChainKeys::keeps_topics(&chain),
+                case.get("topics").is_some()
+            );
+            if let Some(topics) = case.get("topics") {
+                assert_eq!(keys.topics(), topics.as_str().unwrap());
+            }
         }
     }
 
@@ -262,6 +323,7 @@ mod tests {
             let notifications = notifications(prefix, keys.slot());
             assert_eq!(slot(&notifications), slot(&keys.log(topic)));
             assert_eq!(slot(&notifications), slot(&keys.meta(topic)));
+            assert_eq!(slot(&notifications), slot(&keys.topics()));
             assert_eq!(
                 chain_of_log(prefix, &keys.log(topic)),
                 Some((chain, topic.to_string()))
@@ -271,6 +333,8 @@ mod tests {
         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);
+        assert_eq!(chain_of_log("p", "p:{ns:!wa:wf:run}:t:out"), None);
+        assert_eq!(chain_of_log("p", "p:{ns:!a:wf:run:act}:t:out"), None);
     }
 
     #[test]
@@ -345,6 +409,51 @@ mod tests {
         assert_eq!(activity_owner_pattern("p", "ns", "a*b"), "p:{ns:!a:a%2Ab:*");
     }
 
+    #[test]
+    fn a_workflow_activity_chain_keeps_apart_from_its_workflow_and_other_runs() {
+        let wfa = |workflow_id: &str, run_id: &str, activity_id: &str| ChainId {
+            namespace: "ns".to_string(),
+            workflow_id: workflow_id.to_string(),
+            first_run_id: run_id.to_string(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            activity_id: activity_id.to_string(),
+        };
+        let starts = |pattern: String, key: String| key.starts_with(pattern.trim_end_matches('*'));
+        for id in ["wf", "!wa", "wa", "!wa:wf", "!a"] {
+            let keys = ChainKeys::new("p", &wfa(id, "run-1", "1"));
+            for workflow_id in [id, "!wa", "wa", "%21wa", "!a"] {
+                assert!(
+                    !starts(owner_pattern("p", "ns", workflow_id), keys.chain()),
+                    "{id}"
+                );
+                assert!(!starts(
+                    activity_owner_pattern("p", "ns", workflow_id),
+                    keys.chain()
+                ));
+            }
+            assert!(starts(
+                workflow_activity_pattern("p", "ns", id),
+                keys.topics()
+            ));
+            assert!(starts(
+                workflow_activity_pattern("p", "ns", id),
+                keys.log("out")
+            ));
+            // The `:` after the id keeps `wf` from matching the Activities of `wf2`.
+            let longer = ChainKeys::new("p", &wfa(&format!("{id}2"), "run-1", "1"));
+            assert!(!starts(
+                workflow_activity_pattern("p", "ns", id),
+                longer.chain()
+            ));
+            // Activity `1` of the next run is another chain.
+            assert_ne!(keys, ChainKeys::new("p", &wfa(id, "run-2", "1")));
+        }
+        assert_eq!(
+            workflow_activity_pattern("p", "ns", "a*b"),
+            "p:{ns:!wa: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 77077c22da..3e8d9776fa 100644
--- a/crates/streams/src/redis/mod.rs
+++ b/crates/streams/src/redis/mod.rs
@@ -18,7 +18,9 @@ use crate::{
 };
 use conn::{BlockingPool, Shared};
 use errors::{Call, Mapped};
-use keys::{ChainKeys, activity_owner_pattern, owner_pattern, session_field};
+use keys::{
+    ChainKeys, activity_owner_pattern, owner_pattern, session_field, workflow_activity_pattern,
+};
 use prost::Message;
 use redis::{Cmd, Script, Value, streams::StreamReadReply};
 use std::{
@@ -668,6 +670,9 @@ impl StreamStore for RedisStore {
         if self.log.is_some() {
             script.key(keys::notifications(&self.prefix, keys.slot()));
         }
+        if request.chain.as_ref().is_some_and(ChainKeys::keeps_topics) {
+            script.key(keys.topics());
+        }
         script
             .arg(self.retention_ms)
             .arg(self.grace_ms)
@@ -948,6 +953,29 @@ impl StreamStore for RedisStore {
             .mapped(Call::Write)
     }
 
+    async fn written_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
+        if !ChainKeys::keeps_topics(chain) {
+            return Ok(Vec::new());
+        }
+        let keys = ChainKeys::new(&self.prefix, chain);
+        let logs: Vec<String> = redis::cmd("SMEMBERS")
+            .arg(keys.topics())
+            .query_async(&mut self.shared.clone())
+            .await
+            .mapped(Call::Read)?;
+        let mut topics = logs
+            .iter()
+            .map(|log| match keys::chain_of_log(&self.prefix, log) {
+                Some((of, topic)) if of == *chain => Ok(topic),
+                _ => Err(StreamError::storage(format!(
+                    "the topic set of a Workflow Activity holds {log:?}, which is no log key of it"
+                ))),
+            })
+            .collect::<StreamResult<Vec<_>>>()?;
+        topics.sort();
+        Ok(topics)
+    }
+
     async fn list_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
         let keys = ChainKeys::new(&self.prefix, chain);
         let pattern = keys.topics_pattern();
@@ -974,13 +1002,16 @@ impl StreamStore for RedisStore {
     }
 
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
-        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)
-            }
+        let patterns = match StreamOwnerKind::try_from(request.owner_kind) {
+            Ok(StreamOwnerKind::Workflow) => vec![
+                owner_pattern(&self.prefix, &request.namespace, &request.workflow_id),
+                workflow_activity_pattern(&self.prefix, &request.namespace, &request.workflow_id),
+            ],
+            Ok(StreamOwnerKind::Activity) => vec![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 \
@@ -992,12 +1023,17 @@ impl StreamStore for RedisStore {
         // 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 mut deleted = 0;
-        for node in self.shared.primaries().await.mapped(Call::Write)? {
+        let nodes = self.shared.primaries().await.mapped(Call::Write)?;
+        for (node, pattern) in nodes
+            .iter()
+            .flat_map(|node| patterns.iter().map(move |pattern| (node, pattern)))
+            .collect::<Vec<_>>()
+        {
             let mut cursor = "0".to_string();
             loop {
                 let (next, keys) = self
                     .shared
-                    .scan(node.as_ref(), &cursor, &pattern)
+                    .scan(node.as_ref(), &cursor, pattern)
                     .await
                     .mapped(Call::Write)?;
                 for batch in keys.chunks(DELETE_BATCH) {
diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs
index 8bd3cf7f8c..a0387d2a9b 100644
--- a/crates/streams/src/redis/scripts.rs
+++ b/crates/streams/src/redis/scripts.rs
@@ -17,6 +17,10 @@ pub(crate) const APPEND: &str = concat!(
 
 pub(crate) static APPEND_SCRIPT: LazyLock<Script> = LazyLock::new(|| Script::new(APPEND));
 
+/// The line that names the key of a Workflow Activity's topic set, which only its appends pass.
+/// It is the key after the notification stream when the store keeps a notification log.
+const TOPICS_KEY: &str = "local topics = KEYS[5]\n";
+
 /// What a store with a notification log writes after the batch, in place of the append's last
 /// answer: an entry in the chain's notification stream (`KEYS[5]`), and the stream's length, so
 /// Core rings the doorbell when the entry is the stream's only one. A repeat writes no entry.
@@ -25,13 +29,15 @@ const LOGGED_ANSWER: &str = "redis.call('XADD', KEYS[5], 'MAXLEN', '~', 10000, '
 return {first, last, redis.call('XLEN', KEYS[5])}
 ";
 
-/// [APPEND] with its last answer replaced by [LOGGED_ANSWER].
+/// [APPEND] with its last answer replaced by [LOGGED_ANSWER], and the topic set as `KEYS[6]`.
 pub(crate) static APPEND_LOGGED: LazyLock<String> = LazyLock::new(|| {
     let (body, last) = APPEND
         .trim_end()
         .rsplit_once('\n')
         .expect("the append script has a last line");
     assert_eq!(last, "return {first, last}");
+    assert_eq!(body.matches(TOPICS_KEY).count(), 1);
+    let body = body.replacen(TOPICS_KEY, "local topics = KEYS[6]\n", 1);
     format!("{body}\n{LOGGED_ANSWER}")
 });
 
@@ -61,6 +67,15 @@ mod tests {
     /// The close check the append script adds to the SDKs' one.
     const TOPIC_CHECK: &str = " or redis.call('HGET', KEYS[2], 'closed')";
 
+    /// What the append script adds for a Workflow Activity's topic set.
+    const TOPIC_SET: &str = "local topics = KEYS[5]
+if topics then
+  -- On every batch, since the first one can be lost and the close must still find the topic.
+  redis.call('SADD', topics, KEYS[1])
+  redis.call('PEXPIRE', topics, retention + grace)
+end
+";
+
     /// What the floor adds to the append script, each with what it replaces.
     const FLOOR_EDITS: [(&str, &str); 8] = [
         (
@@ -95,15 +110,17 @@ mod tests {
     ];
 
     #[test]
-    fn the_logged_append_differs_in_its_last_answer_only() {
+    fn the_logged_append_differs_in_its_topic_set_key_and_last_answer_only() {
         assert_eq!(
-            APPEND_LOGGED.replacen(LOGGED_ANSWER, "return {first, last}\n", 1),
+            APPEND_LOGGED
+                .replacen(LOGGED_ANSWER, "return {first, last}\n", 1)
+                .replacen("local topics = KEYS[6]\n", TOPICS_KEY, 1),
             APPEND
         );
         assert_eq!(APPEND_LOGGED.matches("XADD', KEYS[5]").count(), 1);
         assert_eq!(
             APPEND_LOGGED_SCRIPT.get_hash(),
-            "5fd994ec419893cc6406fec8166e1ea9796ec692"
+            "e20b57d34060cd9498c9189e7cf1c2a7ae01190c"
         );
     }
 
@@ -111,12 +128,14 @@ mod tests {
     fn the_append_script_is_the_one_before_the_floor_with_the_floor_edits_only() {
         assert_eq!(
             APPEND_SCRIPT.get_hash(),
-            "ad509b51833fb21b5bdc61a87cba7dbdf8f15233"
+            "792ae093a9c09ea8da9078a724eced11ddf5935d"
         );
         let mut before_floor = APPEND
             .strip_prefix(include_str!("../../lua/floor.lua"))
             .unwrap()
             .to_string();
+        assert_eq!(before_floor.matches(TOPIC_SET).count(), 1);
+        before_floor = before_floor.replacen(TOPIC_SET, "", 1);
         for (added, was) in FLOOR_EDITS {
             assert_eq!(before_floor.matches(added).count(), 1, "{added}");
             before_floor = before_floor.replacen(added, was, 1);
diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs
index 5ac53cd3ab..14b0110edb 100644
--- a/crates/streams/src/service.rs
+++ b/crates/streams/src/service.rs
@@ -1309,6 +1309,9 @@ mod tests {
         ) -> StreamResult<()> {
             self.0.retire_producer(c, t, p, a).await
         }
+        async fn written_topics(&self, c: &crate::proto::ChainId) -> StreamResult<Vec<String>> {
+            self.0.written_topics(c).await
+        }
         async fn delete_owner(
             &self,
             r: DeleteOwnerRequest,
diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs
index 71663bc787..77a010da8d 100644
--- a/crates/streams/src/store.rs
+++ b/crates/streams/src/store.rs
@@ -357,7 +357,16 @@ pub trait StreamStore: Send + Sync {
         unsupported("list the topics of a chain")
     }
 
-    /// Deletes every stream of one owner, across all its chains.
+    /// The topics that producers wrote on a Workflow Activity's chain, so the scheduling
+    /// Workflow's Core can close each one when the Activity ends. A topic enters the set with
+    /// every batch, and a topic that no producer wrote is not in it. Other
+    /// chains keep no set and answer empty. A store that keeps no set lists the chain's topics.
+    async fn written_topics(&self, chain: &ChainId) -> StreamResult<Vec<String>> {
+        self.list_topics(chain).await
+    }
+
+    /// Deletes every stream of one owner, across all its chains. A Workflow's delete also deletes
+    /// the streams of every Activity it scheduled.
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
         let _ = request;
         unsupported("delete the streams of an owner")
diff --git a/crates/streams/src/subscribe.rs b/crates/streams/src/subscribe.rs
index cd3f50d350..e39b1254a2 100644
--- a/crates/streams/src/subscribe.rs
+++ b/crates/streams/src/subscribe.rs
@@ -658,6 +658,58 @@ mod tests {
         assert_eq!(setup.server.detached.lock().unwrap()[0].0, activity_chain);
     }
 
+    #[tokio::test]
+    async fn a_workflow_reader_subscribes_to_a_stream_that_a_workflow_activity_owns() {
+        let setup = setup();
+        let tool = |run_id: &str| StreamAddress {
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            run_id: run_id.to_string(),
+            activity_id: "1".to_string(),
+            ..stream()
+        };
+        let tool_chain = |run_id: &str| ChainId {
+            namespace: "ns".to_string(),
+            workflow_id: "owner".to_string(),
+            first_run_id: run_id.to_string(),
+            owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+            activity_id: "1".to_string(),
+        };
+        // Activity `1` of two runs, and the Workflow's own stream, are three streams.
+        for (address, id) in [(tool("run-a"), "1"), (tool("run-b"), "2")] {
+            let request = SubscribeRequest {
+                stream: Some(address),
+                ..subscribe_request(id)
+            };
+            let answer = subscribe(&setup.service, request).await.unwrap();
+            assert!(answer.attached);
+        }
+        assert!(
+            subscribe(&setup.service, subscribe_request("3"))
+                .await
+                .unwrap()
+                .attached
+        );
+        let attached = setup.server.attached.lock().unwrap().clone();
+        assert_eq!(attached[0].chain, tool_chain("run-a"));
+        assert_eq!(attached[1].chain, tool_chain("run-b"));
+        assert_eq!(attached[2].chain, chain("chain-1"));
+        let unsubscribe = UnsubscribeRequest {
+            stream: Some(tool("run-b")),
+            run_id: "reader-run".to_string(),
+            subscription_id: "2".to_string(),
+            ..Default::default()
+        };
+        setup
+            .service
+            .call("Unsubscribe", &unsubscribe.encode_to_vec())
+            .await
+            .unwrap();
+        assert_eq!(
+            setup.server.detached.lock().unwrap()[0].0,
+            tool_chain("run-b")
+        );
+    }
+
     async fn wait(
         setup: &Setup,
         after_counter: i64,
diff --git a/crates/streams/testdata/redis/keys.json b/crates/streams/testdata/redis/keys.json
index 23f823bbd4..40136f5d45 100644
--- a/crates/streams/testdata/redis/keys.json
+++ b/crates/streams/testdata/redis/keys.json
@@ -92,5 +92,57 @@
     "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"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "default",
+    "owner_kind": "workflow_activity",
+    "workflow_id": "agent",
+    "first_run_id": "run-1",
+    "activity_id": "tool-42",
+    "topic": "tokens",
+    "log": "temporal-streams:{default:!wa:agent:run-1:tool-42}:t:tokens",
+    "meta": "temporal-streams:{default:!wa:agent:run-1:tool-42}:t:tokens:meta",
+    "chain": "temporal-streams:{default:!wa:agent:run-1:tool-42}:chain",
+    "topics": "temporal-streams:{default:!wa:agent:run-1:tool-42}:topics"
+  },
+  {
+    "prefix": "p{x}",
+    "namespace": "ns:a",
+    "owner_kind": "workflow_activity",
+    "workflow_id": "wf:1{b}!",
+    "first_run_id": "r:1",
+    "activity_id": "act/1",
+    "topic": "t/1",
+    "log": "p%7Bx%7D:{ns%3Aa:!wa:wf%3A1%7Bb%7D%21:r%3A1:act%2F1}:t:t%2F1",
+    "meta": "p%7Bx%7D:{ns%3Aa:!wa:wf%3A1%7Bb%7D%21:r%3A1:act%2F1}:t:t%2F1:meta",
+    "chain": "p%7Bx%7D:{ns%3Aa:!wa:wf%3A1%7Bb%7D%21:r%3A1:act%2F1}:chain",
+    "topics": "p%7Bx%7D:{ns%3Aa:!wa:wf%3A1%7Bb%7D%21:r%3A1:act%2F1}:topics"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "ns",
+    "owner_kind": "workflow_activity",
+    "workflow_id": "!wa",
+    "first_run_id": "run",
+    "activity_id": "!a",
+    "topic": "out",
+    "log": "temporal-streams:{ns:!wa:%21wa:run:%21a}:t:out",
+    "meta": "temporal-streams:{ns:!wa:%21wa:run:%21a}:t:out:meta",
+    "chain": "temporal-streams:{ns:!wa:%21wa:run:%21a}:chain",
+    "topics": "temporal-streams:{ns:!wa:%21wa:run:%21a}:topics"
+  },
+  {
+    "prefix": "temporal-streams",
+    "namespace": "ns",
+    "owner_kind": "workflow_activity",
+    "workflow_id": "wörk",
+    "first_run_id": "01a1-xyz",
+    "activity_id": "1",
+    "topic": "tópic",
+    "log": "temporal-streams:{ns:!wa:w%C3%B6rk:01a1-xyz:1}:t:t%C3%B3pic",
+    "meta": "temporal-streams:{ns:!wa:w%C3%B6rk:01a1-xyz:1}:t:t%C3%B3pic:meta",
+    "chain": "temporal-streams:{ns:!wa:w%C3%B6rk:01a1-xyz:1}:chain",
+    "topics": "temporal-streams:{ns:!wa:w%C3%B6rk:01a1-xyz:1}:topics"
   }
 ]
diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs
index f5855bdd41..af082d8ba8 100644
--- a/crates/streams/tests/conformance.rs
+++ b/crates/streams/tests/conformance.rs
@@ -168,6 +168,58 @@ impl<C: Case> Case for Activity<C> {
     }
 }
 
+/// Another case whose chains are Workflow Activities, each named by a Workflow, its scheduling
+/// run and an Activity id.
+struct WorkflowActivity<C>(C);
+
+async fn workflow_activity<C: Case>(
+    inner: impl std::future::Future<Output = Option<C>>,
+) -> Option<WorkflowActivity<C>> {
+    Some(WorkflowActivity(inner.await?))
+}
+
+/// The chain of Activity `activity_id` that run `run_id` of `workflow_id` scheduled.
+fn workflow_activity_chain(
+    namespace: &str,
+    workflow_id: &str,
+    run_id: &str,
+    activity_id: &str,
+) -> ChainId {
+    ChainId {
+        namespace: namespace.to_string(),
+        workflow_id: workflow_id.to_string(),
+        first_run_id: run_id.to_string(),
+        owner_kind: StreamOwnerKind::WorkflowActivity as i32,
+        activity_id: activity_id.to_string(),
+    }
+}
+
+#[async_trait::async_trait]
+impl<C: Case> Case for WorkflowActivity<C> {
+    fn store(&self) -> Arc<dyn StreamStore> {
+        self.0.store()
+    }
+
+    fn chain(&self) -> ChainId {
+        let chain = self.0.chain();
+        // SDK default Activity ids repeat in every run, so each chain reuses `1`.
+        workflow_activity_chain(
+            &chain.namespace,
+            &chain.workflow_id,
+            &chain.first_run_id,
+            "1",
+        )
+    }
+
+    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();
@@ -228,16 +280,20 @@ 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)
+        let run = part(&chain.first_run_id);
+        let owner = match StreamOwnerKind::try_from(chain.owner_kind) {
+            Ok(StreamOwnerKind::Activity) => format!("!a:{}:{run}", part(&chain.activity_id)),
+            Ok(StreamOwnerKind::WorkflowActivity) => format!(
+                "!wa:{}:{run}:{}",
+                part(&chain.workflow_id),
+                part(&chain.activity_id)
+            ),
+            _ => format!("{}:{run}", part(&chain.workflow_id)),
         };
         format!(
-            "{}:{{{}:{owner}:{}}}:t:{}",
+            "{}:{{{}:{owner}}}:t:{}",
             part(prefix),
             part(&chain.namespace),
-            part(&chain.first_run_id),
             part(topic)
         )
     }
@@ -591,7 +647,7 @@ fn target(chain: &ChainId, topic: &str) -> ReadTarget {
         topic: topic.to_string(),
         stream_hash: {
             let (kind, id) = chain_owner(chain);
-            stream_hash(&chain.namespace, kind, id, topic)
+            stream_hash(&chain.namespace, kind, &id, topic)
         },
     }
 }
@@ -1497,6 +1553,99 @@ mod cases {
         );
     }
 
+    pub(crate) async fn a_workflow_activity_keeps_the_topics_it_wrote(case: &dyn Case) {
+        let store = case.store();
+        let chain = case.chain();
+        let keeps = chain.owner_kind == StreamOwnerKind::WorkflowActivity as i32;
+        assert!(store.written_topics(&chain).await.unwrap().is_empty());
+        let mut first = Producer::new(case, &chain, OUT, "p", 1);
+        first.append(&["1"]).await;
+        first.append(&["2"]).await;
+        // A repeat of the newest batch adds nothing.
+        first.sequence = 2;
+        first.append(&["2"]).await;
+        Producer::new(case, &chain, OTHER, "q", 2)
+            .append(&["1"])
+            .await;
+        let written = store.written_topics(&chain).await.unwrap();
+        if keeps {
+            assert_eq!(written, [OTHER, OUT]);
+        } else {
+            assert!(written.is_empty());
+        }
+        // A refused batch adds nothing, and a closed topic stays in the set.
+        store.close_topic(&chain, OUT).await.unwrap();
+        let refused = Producer::new(case, &chain, OUT, "r", 1)
+            .send(&[data("3")])
+            .await;
+        assert_eq!(kind(refused), StreamFailureKind::Closed);
+        assert_eq!(store.written_topics(&chain).await.unwrap(), written);
+        // Another chain's set is its own.
+        assert!(
+            store
+                .written_topics(&case.chain())
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    pub(crate) async fn a_workflow_activity_keeps_a_topic_whose_first_batch_was_lost(
+        case: &dyn Case,
+    ) {
+        let chain = case.chain();
+        let mut producer = Producer::new(case, &chain, OUT, "p", 1);
+        producer.sequence = 3;
+        producer.append(&["3"]).await;
+        let written = case.store().written_topics(&chain).await.unwrap();
+        if chain.owner_kind == StreamOwnerKind::WorkflowActivity as i32 {
+            assert_eq!(written, [OUT]);
+        } else {
+            assert!(written.is_empty());
+        }
+    }
+
+    pub(crate) async fn deleting_a_workflow_drops_its_workflow_activities(case: &dyn Case) {
+        let store = case.store();
+        let chain = case.chain();
+        let sibling = ChainId {
+            first_run_id: format!("{}-next", chain.first_run_id),
+            ..chain.clone()
+        };
+        let workflow = ChainId {
+            namespace: chain.namespace.clone(),
+            workflow_id: chain.workflow_id.clone(),
+            first_run_id: chain.first_run_id.clone(),
+            ..Default::default()
+        };
+        // A Workflow whose id only starts with this one keeps its Activities' streams.
+        let bystander = ChainId {
+            workflow_id: format!("{}2", chain.workflow_id),
+            ..chain.clone()
+        };
+        for written in [&chain, &sibling, &workflow, &bystander] {
+            Producer::new(case, written, 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(),
+            ..Default::default()
+        };
+        assert!(store.delete_owner(request.clone()).await.unwrap().deleted > 0);
+        assert_eq!(store.delete_owner(request).await.unwrap().deleted, 0);
+        for gone in [&chain, &sibling, &workflow] {
+            assert_eq!(latest(case, gone, OUT).await, "");
+        }
+        assert!(store.written_topics(&chain).await.unwrap().is_empty());
+        assert_eq!(
+            values(&read(case, &bystander, OUT, BEGINNING).await.unwrap()),
+            ["1"]
+        );
+    }
+
     pub(crate) async fn deleting_an_owner_drops_every_chain_of_it(case: &dyn Case) {
         let store = case.store();
         let chain = case.chain();
@@ -1516,7 +1665,7 @@ mod cases {
                 ..Default::default()
             }
         } else {
-            activity_chain(&chain.namespace, id, &chain.first_run_id)
+            activity_chain(&chain.namespace, &id, &chain.first_run_id)
         };
         let landed = Producer::new(case, &chain, OUT, "p", 1)
             .append(&["1"])
@@ -1869,6 +2018,8 @@ macro_rules! store_cases {
             a_second_close_answers_the_first,
             a_close_with_no_outcome_is_the_first_close_too,
             a_chain_close_keeps_no_outcome,
+            a_workflow_activity_keeps_the_topics_it_wrote,
+            a_workflow_activity_keeps_a_topic_whose_first_batch_was_lost,
             the_declared_capabilities_match_the_calls,
             a_lost_cursor_record_reads_as_expired,
         );
@@ -1904,6 +2055,14 @@ mod activity_memory {
     conformance!(super::activity(super::memory()); deleting_an_owner_drops_every_chain_of_it);
 }
 
+mod workflow_activity_memory {
+    store_cases!(super::workflow_activity(super::memory()));
+    log_cases!(super::workflow_activity(super::memory_with_log()));
+    conformance!(super::workflow_activity(super::memory());
+        deleting_a_workflow_drops_its_workflow_activities,
+    );
+}
+
 mod notifying_memory {
     store_cases!(super::notifying(super::memory()));
     log_cases!(super::notifying(super::memory_with_log()));
@@ -1976,6 +2135,62 @@ mod activity_cluster_redis {
     );
 }
 
+#[cfg(feature = "redis")]
+mod workflow_activity_standalone_redis {
+    store_cases!(super::workflow_activity(super::on_redis::redis(
+        "STREAMS_REDIS_URL",
+        false
+    )));
+    log_cases!(super::workflow_activity(super::on_redis::redis_with_log(
+        "STREAMS_REDIS_URL",
+        false
+    )));
+    conformance!(super::workflow_activity(super::on_redis::redis("STREAMS_REDIS_URL", false));
+        deleting_a_workflow_drops_its_workflow_activities,
+    );
+}
+
+#[cfg(feature = "redis")]
+mod workflow_activity_documented_acl_redis {
+    // The topic set needs `SADD` and `SMEMBERS`, which the page names.
+    store_cases!(super::workflow_activity(
+        super::on_redis::redis_as_documented_user("STREAMS_REDIS_URL", false, &[])
+    ));
+    conformance!(super::workflow_activity(super::on_redis::redis_as_documented_user(
+        "STREAMS_REDIS_URL",
+        false,
+        &["+scan"]
+    ));
+        deleting_a_workflow_drops_its_workflow_activities,
+    );
+}
+
+#[cfg(feature = "redis")]
+mod workflow_activity_documented_acl_cluster_redis_with_log {
+    store_cases!(super::workflow_activity(
+        super::on_redis::redis_as_documented_user_with_log("STREAMS_REDIS_CLUSTER_URL", true)
+    ));
+}
+
+#[cfg(feature = "redis")]
+mod workflow_activity_cluster_redis {
+    store_cases!(super::workflow_activity(super::on_redis::redis(
+        "STREAMS_REDIS_CLUSTER_URL",
+        true
+    )));
+    log_cases!(super::workflow_activity(super::on_redis::redis_with_log(
+        "STREAMS_REDIS_CLUSTER_URL",
+        true
+    )));
+    // Each Activity's chain hashes to its own slot, so the delete reaches every primary.
+    conformance!(super::workflow_activity(super::on_redis::redis(
+        "STREAMS_REDIS_CLUSTER_URL",
+        true
+    ));
+        deleting_a_workflow_drops_its_workflow_activities,
+    );
+}
+
 #[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/redis_store.rs b/crates/streams/tests/redis_store.rs
index 4aac4e51b3..68144dc3e8 100644
--- a/crates/streams/tests/redis_store.rs
+++ b/crates/streams/tests/redis_store.rs
@@ -1247,7 +1247,7 @@ async fn an_evicting_primary_of_a_cluster_is_warned_about() {
 }
 
 #[tokio::test]
-async fn the_documented_acl_is_the_python_sdks() {
+async fn the_documented_acl_is_the_python_sdks_with_cores_additions() {
     // One rule set for every SDK, so an application's ACL keeps working across the move.
     let Ok(checkout) = std::env::var("STREAMS_BROOK_PY") else {
         eprintln!("set STREAMS_BROOK_PY to compare the documented ACL with the Python SDK's");
@@ -1266,8 +1266,13 @@ async fn the_documented_acl_is_the_python_sdks() {
     )
     .unwrap();
     // A cluster client asks the topology with `CLUSTER SLOTS`, which the Python SDK's page left
-    // out. It is the one rule Core adds.
+    // out. Core also adds the set calls of a Workflow Activity's topic set.
     let mut expected = rules(&python);
+    let hscan = expected.iter().position(|rule| rule == "+hscan").unwrap();
+    expected.splice(
+        hscan + 1..hscan + 1,
+        ["+sadd".to_string(), "+smembers".to_string()],
+    );
     expected.push("+cluster|slots".to_string());
     assert_eq!(rules(include_str!("../docs/redis.md")), expected);
 }