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 a6fd6e8e6e..e0bfeb85e2 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
@@ -24,6 +24,9 @@ service StreamService {
   rpc Close(CloseRequest) returns (CloseResponse);
   // Deletes every stream of one owner.
   rpc DeleteOwner(DeleteOwnerRequest) returns (DeleteOwnerResponse);
+  // Drops the dedupe state of one producer attempt on one stream. Safe only once the attempt's
+  // last append returned, since a retry of that append would then land twice.
+  rpc RetireProducer(RetireProducerRequest) returns (RetireProducerResponse);
 }
 
 // Which store holds a process's streams, and how to reach it. Lang builds one store from it and
@@ -210,6 +213,18 @@ message DeleteOwnerResponse {
   uint64 deleted = 1;
 }
 
+message RetireProducerRequest {
+  StreamAddress stream = 1;
+  // The producer attempt whose dedupe state goes. Nothing it wrote is touched.
+  oneof producer {
+    NamedProducer named = 2;
+    ActivityProducer activity = 3;
+  }
+}
+
+// A producer attempt with no dedupe state left, or a stream that holds nothing, answers the same.
+message RetireProducerResponse {}
+
 // One class per way a stream call can fail, so every lang maps the same failure to the same
 // error.
 enum StreamFailureKind {
diff --git a/crates/sdk-core-c-bridge/src/tests/mod.rs b/crates/sdk-core-c-bridge/src/tests/mod.rs
index c5ffb56509..ba49907a18 100644
--- a/crates/sdk-core-c-bridge/src/tests/mod.rs
+++ b/crates/sdk-core-c-bridge/src/tests/mod.rs
@@ -517,7 +517,8 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() {
     use temporalio_common::protos::temporal::api::common::v1::Payload;
     use temporalio_sdk_core::streams::proto::{
         AppendRecord, AppendRequest, AppendResponse, NamedProducer, ReadRequest, ReadResponse,
-        StreamAddress, StreamOwnerKind, StreamRecordKind, append_request, read_record,
+        RetireProducerRequest, StreamAddress, StreamOwnerKind, StreamRecordKind, append_request,
+        read_record, retire_producer_request,
     };
     Context::with(|context| {
         context.runtime_new().unwrap();
@@ -569,7 +570,7 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() {
                 &store,
                 "Read",
                 ReadRequest {
-                    stream: Some(address),
+                    stream: Some(address.clone()),
                     ..Default::default()
                 }
                 .encode_to_vec(),
@@ -585,6 +586,24 @@ fn test_stream_store_appends_and_reads_through_the_c_bridge() {
             panic!("expected a stored record, got {record:?}");
         };
         assert_eq!(stored.body.as_ref().unwrap().data, b"encrypted");
+
+        // The producer's last append returned, so its dedupe state may go.
+        let retired = context
+            .stream_store_call(
+                &store,
+                "RetireProducer",
+                RetireProducerRequest {
+                    stream: Some(address),
+                    producer: Some(retire_producer_request::Producer::Named(NamedProducer {
+                        producer_id: "p".to_string(),
+                        attempt: 1,
+                    })),
+                }
+                .encode_to_vec(),
+            )
+            .unwrap()
+            .unwrap();
+        assert!(retired.is_empty());
     });
 }
 
diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md
index 9aedcb07cd..d7faf8a18e 100644
--- a/crates/sdk-core/CHANGELOG.md
+++ b/crates/sdk-core/CHANGELOG.md
@@ -40,6 +40,11 @@ relevant information.
   requests.
 * Experimental: the C bridge has `temporal_core_stream_store_new`, `_call` and `_free` for the
   stream service, built with the Redis store.
+* Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe
+  state on a stream, which the Redis store keeps as one field of the topic's meta hash until
+  retention. A producer that writes once per short-lived attempt calls it after the attempt's
+  last append returned, so the meta hash tracks the attempts in flight. Calling it earlier lets a
+  retried batch land twice.
 
 ### Breaking Changes
 * Replay detects changes to the service or operation of scheduled Nexus operations, reporting
diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md
index 44e2128292..08193a4177 100644
--- a/crates/streams/docs/redis.md
+++ b/crates/streams/docs/redis.md
@@ -12,6 +12,14 @@ count, its first and last entry ids, and lang's digest of it in hex. The script
 new batch once the chain or the topic is closed, in the same step as the write, so no batch
 lands after a close returned. Only that topic check differs from the SDKs' script.
 
+**Retired producers.** A field stays until a write's trim finds it older than the retention, so
+a topic written by many short-lived producers holds one field per producer attempt of the last
+retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with
+`HDEL` and touches nothing else. A missing field or meta is no failure. Call it only once the
+attempt's last append returned: a retry of an earlier batch would find no field to compare with
+and land a second time. Readers lose nothing, since a store compares a batch with its own
+attempt's field only and readers find a newer attempt from the records.
+
 **Retention.** Every script that writes a log trims it with `XTRIM MINID` to the retention and
 slides the expiry of the log and its meta, so a stream dies a retention after its last write
 and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as
diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs
index 036b851db0..d316fecb6f 100644
--- a/crates/streams/src/handle.rs
+++ b/crates/streams/src/handle.rs
@@ -8,10 +8,11 @@ use crate::{
     check_append_digest, mint_cursor,
     owner::{OwnerClient, OwnerDescription, OwnerError},
     proto::{
-        AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, DeleteOwnerRequest,
-        DeleteOwnerResponse, LatestRequest, LatestResponse, ReadRequest, ReadResponse,
-        StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind,
-        append_request::Producer,
+        ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse,
+        DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer,
+        ReadRequest, ReadResponse, RetireProducerRequest, RetireProducerResponse,
+        StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind, append_request,
+        retire_producer_request,
     },
     reader::{ReadState, read_with},
     stored_append_record, stream_hash,
@@ -181,26 +182,14 @@ impl Streams {
     pub async fn append(&self, request: AppendRequest) -> StreamResult<AppendResponse> {
         let stream = address(request.stream.as_ref())?;
         let (producer_id, attempt) = match &request.producer {
-            Some(Producer::Named(named)) => {
-                if named.producer_id.is_empty() {
-                    return Err(StreamError::invalid_argument("a producer needs an id"));
-                }
-                (named.producer_id.clone(), named.attempt)
-            }
-            Some(Producer::Activity(activity)) => {
-                (activity_producer_id(activity), i64::from(activity.attempt))
-            }
+            Some(append_request::Producer::Named(named)) => producer(Some(named), None)?,
+            Some(append_request::Producer::Activity(activity)) => producer(None, Some(activity))?,
             None => {
                 return Err(StreamError::invalid_argument(
                     "an append needs its producer",
                 ));
             }
         };
-        if attempt < 1 {
-            return Err(StreamError::invalid_argument(format!(
-                "a producer's attempt starts at one, not {attempt}"
-            )));
-        }
         if request.sequence < 1 {
             return Err(StreamError::invalid_argument(format!(
                 "a producer's sequence starts at one, not {}",
@@ -214,11 +203,7 @@ impl Streams {
         }
         check_append_digest(&request.digest)?;
         let chain = self
-            .producer_chain(ProducerKey {
-                owner: owner_key(stream),
-                producer_id: producer_id.clone(),
-                attempt,
-            })
+            .producer_chain(producer_key(owner_key(stream), &producer_id, attempt))
             .await?;
         let records = request
             .records
@@ -389,6 +374,40 @@ impl Streams {
         Ok(CloseResponse {})
     }
 
+    /// Drops one producer attempt's dedupe state on a stream, on the chain that attempt wrote
+    /// on. It refuses nothing on a closed stream, and a stream or attempt with no state answers
+    /// the same.
+    ///
+    /// Only safe once the attempt's last append returned: a retry of an earlier batch would land
+    /// a second time. Nothing else is lost. A store compares a batch with its own attempt's
+    /// state only, and readers find a newer attempt from the records.
+    pub async fn retire_producer(
+        &self,
+        request: RetireProducerRequest,
+    ) -> StreamResult<RetireProducerResponse> {
+        let stream = address(request.stream.as_ref())?;
+        let (producer_id, attempt) = match &request.producer {
+            Some(retire_producer_request::Producer::Named(named)) => producer(Some(named), None)?,
+            Some(retire_producer_request::Producer::Activity(activity)) => {
+                producer(None, Some(activity))?
+            }
+            None => {
+                return Err(StreamError::invalid_argument("a retire needs its producer"));
+            }
+        };
+        let key = producer_key(owner_key(stream), &producer_id, attempt);
+        // The chain the attempt's appends resolved, even when a newer chain took the Workflow
+        // id since, so the state goes from where it was written.
+        let chain = match self.producers.get(&key) {
+            Some(checked) => chain_of_key(&key.owner, &checked.first_run_id),
+            None => self.chain(stream).await?,
+        };
+        self.store
+            .retire_producer(&chain, &stream.topic, &key.producer_id, attempt)
+            .await?;
+        Ok(RetireProducerResponse {})
+    }
+
     /// Deletes every stream of one owner, across all its chains. It asks Temporal nothing, so it
     /// works after the owner is gone. Delete only after the chain closed and its producers
     /// stopped, since a late append writes a fresh stream with no dedupe state and no close mark.
@@ -593,6 +612,59 @@ fn address(stream: Option<&StreamAddress>) -> StreamResult<&StreamAddress> {
 }
 
 /// Refuses a topic no store keys reliably, by the rules Core's Workflow path also applies.
+/// The producer id and attempt a named producer or an Activity attempt writes as.
+fn producer(
+    named: Option<&NamedProducer>,
+    activity: Option<&ActivityProducer>,
+) -> StreamResult<(String, i64)> {
+    let (producer_id, attempt) = match (named, activity) {
+        (Some(named), _) => {
+            if named.producer_id.is_empty() {
+                return Err(StreamError::invalid_argument("a producer needs an id"));
+            }
+            // A Workflow run writes its own output as `@<run id>`. A named producer with that id
+            // would share the run's dedupe session and could withdraw its records.
+            if named.producer_id.starts_with('@') {
+                return Err(StreamError::invalid_argument(format!(
+                    "a named producer's id can't start with `@`, which names a Workflow run's \
+                     own output, not {:?}",
+                    named.producer_id
+                )));
+            }
+            (named.producer_id.clone(), named.attempt)
+        }
+        (None, Some(activity)) => (activity_producer_id(activity), i64::from(activity.attempt)),
+        (None, None) => return Err(StreamError::invalid_argument("a call needs its producer")),
+    };
+    if attempt < 1 {
+        return Err(StreamError::invalid_argument(format!(
+            "a producer's attempt starts at one, not {attempt}"
+        )));
+    }
+    Ok((producer_id, attempt))
+}
+
+/// The key a producer attempt's chain is remembered by. A Workflow writes through a new local
+/// producer per flush, `__temporal_<kind>_<n>@<run id>`, and every flush of a run shares the
+/// run's chain. So they share one entry, and a busy run can't push the long-lived producers out.
+fn producer_key(owner: OwnerKey, producer_id: &str, attempt: i64) -> ProducerKey {
+    match producer_id
+        .strip_prefix("__temporal_")
+        .and_then(|rest| rest.rsplit_once('@'))
+    {
+        Some((_, run_id)) => ProducerKey {
+            owner,
+            producer_id: format!("__temporal_@{run_id}"),
+            attempt: 0,
+        },
+        None => ProducerKey {
+            owner,
+            producer_id: producer_id.to_string(),
+            attempt,
+        },
+    }
+}
+
 fn check_topic(topic: &str) -> StreamResult<()> {
     temporalio_common::streams::check_topic(topic).map_err(StreamError::invalid_argument)
 }
@@ -674,7 +746,7 @@ mod tests {
         MemoryStore,
         proto::{
             ActivityProducer, AppendRecord, NamedProducer, StoreReadRequest, StreamFailureKind,
-            StreamRecordKind,
+            StreamRecordKind, append_request::Producer,
         },
     };
     use sha2::{Digest, Sha256};
@@ -1113,6 +1185,41 @@ mod tests {
         assert!(chain_closed(&setup.store, "run-1").await);
     }
 
+    #[tokio::test]
+    async fn a_workflows_local_producers_share_one_memory_entry() {
+        let setup = setup(StreamsOptions {
+            remembered: 2,
+            ..StreamsOptions::default()
+        });
+        setup
+            .streams
+            .append(append("long-lived", 1, "1"))
+            .await
+            .unwrap();
+        for flush in 1..=20 {
+            let producer = format!("__temporal_local_{flush}@run-1");
+            setup
+                .streams
+                .append(append(&producer, 1, "x"))
+                .await
+                .unwrap();
+        }
+        let key = producer_key(owner_key(&stream("out")), "long-lived", 1);
+        assert!(setup.streams.producers.get(&key).is_some());
+    }
+
+    #[tokio::test]
+    async fn a_named_producer_can_not_write_as_a_workflow_run() {
+        let setup = setup(StreamsOptions::default());
+        let error = setup
+            .streams
+            .append(append("@run-1", 1, "1"))
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind, StreamFailureKind::InvalidArgument);
+        assert!(error.message.contains("@run-1"), "{error}");
+    }
+
     #[tokio::test(start_paused = true)]
     async fn one_not_found_from_the_owner_check_closes_nothing() {
         let setup = setup(StreamsOptions::default());
@@ -1266,6 +1373,51 @@ mod tests {
         setup.streams.append(other).await.unwrap();
     }
 
+    #[tokio::test]
+    async fn a_retire_drops_the_attempts_state_on_the_chain_it_wrote() {
+        let setup = setup(StreamsOptions::default());
+        setup.streams.append(append("p", 1, "1")).await.unwrap();
+        setup
+            .streams
+            .close(CloseRequest {
+                stream: Some(stream("out")),
+            })
+            .await
+            .unwrap();
+        let before = setup.owners.describes();
+        let retire = |producer: Option<retire_producer_request::Producer>| RetireProducerRequest {
+            stream: Some(stream("out")),
+            producer,
+        };
+        let named = |attempt| {
+            Some(retire_producer_request::Producer::Named(NamedProducer {
+                producer_id: "p".to_string(),
+                attempt,
+            }))
+        };
+        // A closed stream refuses no retire, and a repeat answers the same.
+        for _ in 0..2 {
+            setup
+                .streams
+                .retire_producer(retire(named(1)))
+                .await
+                .unwrap();
+        }
+        // The append resolved the chain, so the retire asks Temporal nothing.
+        assert_eq!(setup.owners.describes(), before);
+        // With the state gone, the repeat is a new batch, which the closed stream refuses.
+        assert_eq!(
+            kind(setup.streams.append(append("p", 1, "1")).await),
+            StreamFailureKind::Closed
+        );
+        for bad in [retire(None), retire(named(0))] {
+            assert_eq!(
+                kind(setup.streams.retire_producer(bad).await),
+                StreamFailureKind::InvalidArgument
+            );
+        }
+    }
+
     fn quick() -> StreamsOptions {
         StreamsOptions {
             owner_check_min: Duration::from_millis(20),
diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs
index a190c3bfe7..d92ed331e5 100644
--- a/crates/streams/src/memory.rs
+++ b/crates/streams/src/memory.rs
@@ -361,6 +361,23 @@ impl StreamStore for MemoryStore {
         Ok(())
     }
 
+    async fn retire_producer(
+        &self,
+        chain: &ChainId,
+        topic: &str,
+        producer_id: &str,
+        attempt: i64,
+    ) -> StreamResult<()> {
+        let mut chains = self.chains.lock().unwrap();
+        if let Some(topic) = chains
+            .get_mut(&chain_key(chain))
+            .and_then(|chain| chain.topics.get_mut(topic))
+        {
+            topic.held.remove(&(producer_id.to_string(), attempt));
+        }
+        Ok(())
+    }
+
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
         if request.owner_kind != StreamOwnerKind::Workflow as i32 {
             return Err(StreamError::unsupported(format!(
diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs
index 48bf3ad6da..c70ea74921 100644
--- a/crates/streams/src/reader.rs
+++ b/crates/streams/src/reader.rs
@@ -460,6 +460,10 @@ mod tests {
             unreachable!()
         }
 
+        async fn retire_producer(&self, _: &ChainId, _: &str, _: &str, _: i64) -> StreamResult<()> {
+            unreachable!()
+        }
+
         async fn delete_owner(&self, _: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
             unreachable!()
         }
diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs
index 9a1cd0db7a..2c02485e24 100644
--- a/crates/streams/src/redis/mod.rs
+++ b/crates/streams/src/redis/mod.rs
@@ -618,6 +618,23 @@ impl StreamStore for RedisStore {
             .mapped(Call::Write)
     }
 
+    async fn retire_producer(
+        &self,
+        chain: &ChainId,
+        topic: &str,
+        producer_id: &str,
+        attempt: i64,
+    ) -> StreamResult<()> {
+        // One field of one key, so a plain `HDEL` is atomic, and it creates nothing when the meta
+        // is gone.
+        redis::cmd("HDEL")
+            .arg(ChainKeys::new(&self.prefix, chain).meta(topic))
+            .arg(session_field(producer_id, attempt))
+            .query_async::<()>(&mut self.shared.clone())
+            .await
+            .mapped(Call::Write)
+    }
+
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse> {
         if request.owner_kind != StreamOwnerKind::Workflow as i32 {
             return Err(StreamError::unsupported(format!(
diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs
index f4689e3189..e3a695ac49 100644
--- a/crates/streams/src/service.rs
+++ b/crates/streams/src/service.rs
@@ -3,8 +3,8 @@
 use crate::{
     OwnerClient, StreamError, StreamResult, StreamStore, Streams, StreamsOptions,
     proto::{
-        AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest, StreamFailure,
-        StreamStoreConfig, stream_store_config,
+        AppendRequest, CloseRequest, DeleteOwnerRequest, LatestRequest, ReadRequest,
+        RetireProducerRequest, StreamFailure, StreamStoreConfig, stream_store_config,
     },
 };
 use prost::Message;
@@ -98,6 +98,11 @@ impl StreamService {
                 .delete_owner(decode::<DeleteOwnerRequest>(rpc, request)?)
                 .await
                 .map(|r| r.encode_to_vec()),
+            "RetireProducer" => self
+                .streams
+                .retire_producer(decode::<RetireProducerRequest>(rpc, request)?)
+                .await
+                .map(|r| r.encode_to_vec()),
             other => Err(StreamError::unsupported(format!(
                 "the stream service has no call {other:?}"
             ))),
@@ -380,6 +385,42 @@ mod tests {
         assert!(garbled.message.contains("did not decode"));
     }
 
+    #[tokio::test]
+    async fn a_retired_producer_writes_its_batch_again() {
+        let service = service().await;
+        let first = AppendResponse::decode(
+            service
+                .call("Append", &append(1, &["a"]))
+                .await
+                .unwrap()
+                .as_slice(),
+        )
+        .unwrap();
+        let retire = crate::proto::RetireProducerRequest {
+            stream: Some(address()),
+            producer: Some(crate::proto::retire_producer_request::Producer::Named(
+                NamedProducer {
+                    producer_id: "p".to_string(),
+                    attempt: 1,
+                },
+            )),
+        };
+        let answer = service
+            .call("RetireProducer", &retire.encode_to_vec())
+            .await
+            .unwrap();
+        assert!(answer.is_empty());
+        let again = AppendResponse::decode(
+            service
+                .call("Append", &append(1, &["a"]))
+                .await
+                .unwrap()
+                .as_slice(),
+        )
+        .unwrap();
+        assert_ne!(again.first_cursor, first.first_cursor);
+    }
+
     #[tokio::test]
     async fn a_delete_is_checked_before_it_reaches_the_store() {
         let service = service().await;
diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs
index 36bfcd1dae..69b7a2d9f8 100644
--- a/crates/streams/src/store.rs
+++ b/crates/streams/src/store.rs
@@ -75,6 +75,19 @@ pub trait StreamStore: Send + Sync {
     /// Marks one topic of a chain closed, like [StreamStore::close_chain] for that topic alone.
     async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()>;
 
+    /// Drops the newest-batch record [StreamStore::append] keeps for one producer attempt on a
+    /// topic, and nothing else. A missing attempt or topic is no failure.
+    ///
+    /// Call it only once the attempt's last append returned. A retry of an earlier batch would
+    /// then find no record to compare with and land a second time.
+    async fn retire_producer(
+        &self,
+        chain: &ChainId,
+        topic: &str,
+        producer_id: &str,
+        attempt: i64,
+    ) -> StreamResult<()>;
+
     /// Deletes every stream of one owner, across all its chains.
     async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult<DeleteOwnerResponse>;
 }
diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs
index 391f533e21..cd54a18c2d 100644
--- a/crates/streams/tests/conformance.rs
+++ b/crates/streams/tests/conformance.rs
@@ -1122,6 +1122,45 @@ mod cases {
         }
     }
 
+    pub(crate) async fn retiring_a_producer_drops_only_its_dedupe_state(case: &dyn Case) {
+        let store = case.store();
+        let chain = case.chain();
+        let first = Producer::new(case, &chain, OUT, "p", 1)
+            .append(&["1"])
+            .await;
+        let kept = Producer::new(case, &chain, OUT, "q", 1)
+            .append(&["2"])
+            .await;
+        store.retire_producer(&chain, OUT, "p", 1).await.unwrap();
+        // A repeat, an attempt that never wrote, a topic and a chain that hold nothing.
+        store.retire_producer(&chain, OUT, "p", 1).await.unwrap();
+        store.retire_producer(&chain, OUT, "p", 2).await.unwrap();
+        store.retire_producer(&chain, OTHER, "p", 1).await.unwrap();
+        store
+            .retire_producer(&case.chain(), OUT, "p", 1)
+            .await
+            .unwrap();
+        assert_eq!(
+            values(&read(case, &chain, OUT, BEGINNING).await.unwrap()),
+            ["1", "2"]
+        );
+        // The other producer still dedupes.
+        let repeat = Producer::new(case, &chain, OUT, "q", 1)
+            .append(&["2"])
+            .await;
+        assert_eq!(repeat, kept);
+        // The retired attempt no longer does, which is why only an attempt whose last append
+        // returned may be retired.
+        let again = Producer::new(case, &chain, OUT, "p", 1)
+            .append(&["1"])
+            .await;
+        assert_ne!(again.first_position, first.first_position);
+        assert_eq!(
+            values(&read(case, &chain, OUT, BEGINNING).await.unwrap()),
+            ["1", "2", "1"]
+        );
+    }
+
     pub(crate) async fn deleting_an_owner_drops_every_chain_of_it(case: &dyn Case) {
         let store = case.store();
         let chain = case.chain();
@@ -1204,6 +1243,7 @@ mod memory {
         a_closed_topic_leaves_the_others_open,
         a_closed_topic_answers_a_repeat_and_ends_reads,
         a_refused_append_leaves_no_trace_of_its_topic,
+        retiring_a_producer_drops_only_its_dedupe_state,
         deleting_an_owner_drops_every_chain_of_it,
     );
 }
@@ -1245,6 +1285,7 @@ macro_rules! redis_cases {
             a_closed_chain_refuses_new_batches_but_answers_a_repeat,
             a_closed_topic_leaves_the_others_open,
             a_closed_topic_answers_a_repeat_and_ends_reads,
+            retiring_a_producer_drops_only_its_dedupe_state,
         );
     };
 }
diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs
index 94fb45ddae..3ff150e9a6 100644
--- a/crates/streams/tests/redis_store.rs
+++ b/crates/streams/tests/redis_store.rs
@@ -187,6 +187,46 @@ async fn a_batch_is_one_script_and_keeps_one_high_water_field() {
     assert_eq!(parts[4], "05");
 }
 
+#[tokio::test]
+async fn retiring_a_producer_deletes_its_session_field_only() {
+    let Some(mut setup) = setup().await else {
+        return;
+    };
+    let chain = chain();
+    for producer in ["p", "q"] {
+        setup
+            .store
+            .append(append(&chain, producer, 1, 1, b"\x01"))
+            .await
+            .unwrap();
+    }
+    let meta = format!("{}:meta", log_key(&setup.prefix, &chain, "events"));
+    setup
+        .store
+        .retire_producer(&chain, "events", "p", 1)
+        .await
+        .unwrap();
+    let fields: HashMap<String, String> = setup.raw.hgetall(&meta).await.unwrap();
+    assert!(!fields.contains_key("hw:1:p:1"), "{fields:?}");
+    assert!(fields.contains_key("hw:1:q:1"), "{fields:?}");
+    assert_eq!(fields["added"], "2");
+    // A topic Redis holds nothing for keeps holding nothing.
+    setup
+        .store
+        .retire_producer(&chain, "unknown", "p", 1)
+        .await
+        .unwrap();
+    let exists: bool = setup
+        .raw
+        .exists(format!(
+            "{}:meta",
+            log_key(&setup.prefix, &chain, "unknown")
+        ))
+        .await
+        .unwrap();
+    assert!(!exists);
+}
+
 #[tokio::test]
 async fn appends_answer_as_the_python_sdk_recorded() {
     // The outcomes the Python provider's script gave, run again through the Core store.