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 63bca5f6c..7385028a1 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
@@ -29,27 +29,6 @@ message StreamRef {
     string backend = 3;
 }
 
-// Experimental. Might change in the future.
-// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the
-// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs,
-// and which earlier Records a higher epoch withdraws.
-message ProducerEpoch {
-    oneof kind {
-        // A Workflow run, which writes in its Workflow Task. A higher epoch withdraws the Records
-        // of the run's Workflow Tasks that never completed, also those of their Local Activities.
-        WorkflowTaskEpoch workflow = 1;
-        // A Local Activity. A higher epoch withdraws the Producer's lower epochs.
-        LocalActivityEpoch local_activity = 2;
-        // An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs.
-        ActivityEpoch activity = 3;
-        // A Standalone Activity. A higher epoch withdraws the Producer's lower epochs.
-        ActivityEpoch standalone_activity = 4;
-        // A client. A higher epoch only fences the lower ones, since a restarted client's earlier
-        // Records are real messages, not the Records of a failed attempt.
-        ClientEpoch client = 5;
-    }
-}
-
 // Experimental. Might change in the future.
 // One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then
 // started_time.
@@ -86,32 +65,78 @@ message ClientEpoch {
 }
 
 // Experimental. Might change in the future.
-// The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them.
+// A Workflow run Producer. It writes in its Workflow Task. A higher epoch withdraws the Records of
+// the run's Workflow Tasks that never completed, with those of their Local Activities.
+message WorkflowProducer {
+    // The first execution run id of the run chain.
+    string run_id = 1;
+    WorkflowTaskEpoch epoch = 2;
+}
+
+// Experimental. Might change in the future.
+// A Local Activity Producer. A higher epoch withdraws the Producer's lower epochs.
+message LocalActivityProducer {
+    string run_id = 1;
+    LocalActivityEpoch epoch = 2;
+}
+
+// Experimental. Might change in the future.
+// An Activity Producer, for an Activity of a Workflow or a Standalone Activity. A higher epoch
+// withdraws the Producer's lower epochs.
+message ActivityProducer {
+    string run_id = 1;
+    ActivityEpoch epoch = 2;
+}
+
+// Experimental. Might change in the future.
+// A client Producer. A higher epoch only fences the lower ones, since a restarted client's earlier
+// Records are real messages, not the Records of a failed attempt. A client has no run.
+message ClientProducer {
+    ClientEpoch epoch = 1;
+}
+
+// Experimental. Might change in the future.
+// The code that wrote a batch. Its id stays the same across attempts, and the kind's case orders
+// its epochs and picks which earlier Records a higher epoch withdraws. A run-based kind carries its
+// run, so a client can't.
 message Producer {
     // Stable across attempts, such as `wf:<run id>` for the Workflow Producer of a run.
     string id = 1;
-    // Required. Its case gives the kind of the Producer.
-    ProducerEpoch epoch = 2;
-    // The run of the Producer, so nobody parses the id. Empty for a client.
-    string run_id = 3;
+    // Required. The case is the kind of the Producer.
+    oneof kind {
+        WorkflowProducer workflow = 2;
+        LocalActivityProducer local_activity = 3;
+        ActivityProducer activity = 4;
+        ActivityProducer standalone_activity = 5;
+        ClientProducer client = 6;
+    }
 }
 
 // Experimental. Might change in the future.
 // The Producer of a write, and which earlier Records the write withdraws. An append, a close and
 // a writer carry the same fields, so the Store orders and withdraws each write the same way.
 message StreamWriter {
-    // Required on a write. Its epoch's case picks the withdrawal rule.
+    // Required on a write. Its Producer's kind picks the withdrawal rule.
     Producer producer = 1;
-    // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local
-    // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's
-    // Local Activity Records whose Workflow Task never completed.
-    WorkflowTaskEpoch previous_completed_task = 2;
-    // Workflow and Local Activity only. The started event ids of the run's failed or timed out
-    // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks.
-    repeated int64 failed_tasks = 3;
-    // Retry run only. The previous run, on the retry run's first write. Every Record whose
-    // Producer carries it is withdrawn.
-    string withdraw_run = 4;
+    // Set only by a Workflow or Local Activity write, since only a run withdraws orphaned Local
+    // Activity Records or a previous run. Unset for an Activity or client write.
+    RunWithdrawal withdrawal = 2;
+}
+
+// Experimental. Might change in the future.
+// What a Workflow or Local Activity write withdraws from its run. One write can carry both the task
+// facts and a run to withdraw, so these are fields of one message and not a oneof.
+message RunWithdrawal {
+    // The run's last completed Workflow Task, from a Local Activity's start. Its started_event_id
+    // is 0 before the first. The write withdraws the run's Local Activity Records whose Workflow
+    // Task never completed.
+    WorkflowTaskEpoch previous_completed_task = 1;
+    // The started event ids of the run's failed or timed out Workflow Tasks. The write withdraws
+    // the run's Local Activity Records of these tasks.
+    repeated int64 failed_tasks = 2;
+    // Retry run only. The previous run, on the retry run's first write. Every Record whose Producer
+    // carries it is withdrawn.
+    string withdraw_run = 3;
 }
 
 // Experimental. Might change in the future.
@@ -125,8 +150,7 @@ message AppendRecord {
 // batch of a higher epoch, or for a retry run's first write, and a read builds a Superseded item
 // from them. A reader drops the earlier Records that the facts match.
 message StreamSuperseded {
-    // The writer of the batch that withdraws. Its Producer's epoch is empty when only
-    // withdraw_run applies.
+    // The writer of the batch that withdraws, with its withdrawal facts.
     StreamWriter writer = 1;
 }
 
diff --git a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto
index cdd61456e..463f91194 100644
--- a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto
+++ b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto
@@ -75,7 +75,7 @@ message AppendRecordsRequest {
     int64 sequence = 4;
     repeated temporal.api.stream.v1.AppendRecord records = 5;
     // Only read by the server, on a Workflow Producer's batch. The batch withdraws no Workflow
-    // Task and keeps no task facts, but the writer's withdraw_run still applies.
+    // Task and keeps no task facts, but the writer's withdrawal withdraw_run still applies.
     bool defer_withdrawal = 6;
     // Only read by the server. The server checks the Producer against the Activity's current
     // attempt, so nobody else writes as that Activity.
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 0c9b198b7..050ac393a 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
@@ -20,9 +20,10 @@ import "temporal/api/stream/v1/message.proto";
 message OpenWriterRequest {
     string namespace = 1;
     temporal.api.stream.v1.StreamRef ref = 2;
-    // Each batch sends it as AppendRecords does. For a Local Activity it has the run's task facts,
-    // so the writes withdraw the run's orphaned Local Activity Records. Its withdraw_run must be
-    // empty, since only a retry run's first append withdraws a run, and a writer sends many.
+    // Each batch sends it as AppendRecords does. For a Local Activity its withdrawal has the run's
+    // task facts, so the writes withdraw the run's orphaned Local Activity Records. The withdrawal's
+    // withdraw_run must be empty, since only a retry run's first append withdraws a run, and a
+    // writer sends many.
     temporal.api.stream.v1.StreamWriter writer = 3;
     // A write sends at most this many Records in one batch. Zero is 1000.
     int32 max_batch_size = 4;
diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml
index ec470a62f..eefe30a42 100644
--- a/crates/streams/Cargo.toml
+++ b/crates/streams/Cargo.toml
@@ -12,13 +12,16 @@ keywords = ["temporal", "workflow", "streams"]
 categories = ["development-tools"]
 
 [features]
-# The test helpers, for the crates of other Stores.
-test-support = []
+# The shared suites, for the crates of other Stores.
+test-support = ["dep:futures-util", "dep:serde_json"]
 
 [dependencies]
 async-trait = "0.1"
+futures-util = { version = "0.3", default-features = false, features = ["std"], optional = true }
+parking_lot = "0.12"
 prost = { workspace = true }
 prost-types = { workspace = true }
+serde_json = { workspace = true, optional = true }
 sha2 = "0.10"
 thiserror = { workspace = true }
 tokio = { version = "1.47", default-features = false, features = ["rt", "sync", "time"] }
@@ -30,7 +33,9 @@ version = "~1.1.0"
 default-features = false
 
 [dev-dependencies]
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
 serde_json = { workspace = true }
+tokio = { version = "1.47", features = ["macros", "rt-multi-thread", "time"] }
 
 [lints]
 workspace = true
diff --git a/crates/streams/README.md b/crates/streams/README.md
index c53d50b60..634a43b22 100644
--- a/crates/streams/README.md
+++ b/crates/streams/README.md
@@ -17,5 +17,14 @@ re-exports them, so a value needs no conversion between the crates.
 - `testdata/*.json` are the shared vectors. They cover Stream ids, Producer ids, Store keys and
   meta fields, offsets, epoch keys, the withdrawal rule and the batch digest of a dedupe entry.
   `testdata/gen_vectors.py` writes them from a second implementation, and the crate's tests read
-  each file. The Python SDK also reads `ids.json` and `withdrawals.json`. `digests.json` is the
-  golden of the crate's batch digest.
+  each file. The Python SDK also reads `ids.json` and `withdrawals.json`. Each Store reads
+  `fields.json` through the shared Store suite. `digests.json` is the golden of the crate's batch
+  digest.
+
+## Stores and the shared suites
+
+The memory Store is for tests. It keeps the key layout of the Redis Store and applies the same rules
+in the same order. The `test-support` feature exports the shared suites in `testing`. A Store crate
+implements `testing::TestBackend` for its Store and runs each suite with `testing::run`, so that
+every Store keeps one set of rules. This crate runs them on the memory Store in its unit tests.
+
diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs
index 697d66ef6..15b1ca6a2 100644
--- a/crates/streams/src/epoch.rs
+++ b/crates/streams/src/epoch.rs
@@ -7,8 +7,8 @@
 use crate::{
     StreamError, StreamResult,
     proto::{
-        Producer, ProducerEpoch, StreamRecord, StreamSuperseded, WorkflowTaskEpoch,
-        producer_epoch::Kind, stream_record::Item,
+        Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind,
+        stream_record::Item,
     },
 };
 use prost_types::Timestamp;
@@ -42,22 +42,52 @@ impl From<&Kind> for ProducerKind {
     }
 }
 
-/// Refuses a Producer with no epoch, or with an epoch that has no case.
+/// Refuses a Producer with no kind case.
 impl TryFrom<&Producer> for ProducerKind {
     type Error = StreamError;
 
     fn try_from(producer: &Producer) -> StreamResult<Self> {
-        let epoch = producer
-            .epoch
-            .as_ref()
-            .ok_or_else(|| StreamError::invalid("The Producer has no epoch"))?;
-        kind(epoch).ok_or_else(|| StreamError::invalid("The Producer's epoch has no kind"))
+        kind_of(producer).ok_or_else(|| StreamError::invalid("The Producer has no kind"))
     }
 }
 
-/// The kind of an epoch's case, `None` when the epoch has no case.
-fn kind(epoch: &ProducerEpoch) -> Option<ProducerKind> {
-    epoch.kind.as_ref().map(ProducerKind::from)
+/// The kind case of a Producer, `None` when it has none.
+pub fn kind_of(producer: &Producer) -> Option<ProducerKind> {
+    producer.kind.as_ref().map(ProducerKind::from)
+}
+
+/// The run of a Producer. Empty for a client, which has no run.
+pub fn producer_run_id(producer: &Producer) -> &str {
+    match &producer.kind {
+        Some(Kind::Workflow(w)) => &w.run_id,
+        Some(Kind::LocalActivity(l)) => &l.run_id,
+        Some(Kind::Activity(a) | Kind::StandaloneActivity(a)) => &a.run_id,
+        Some(Kind::Client(_)) | None => "",
+    }
+}
+
+/// The run that a writer withdraws, empty for none.
+pub fn writer_withdraw_run(writer: &StreamWriter) -> &str {
+    writer
+        .withdrawal
+        .as_ref()
+        .map_or("", |w| w.withdraw_run.as_str())
+}
+
+/// The started event ids of the run's failed Workflow Tasks that a writer reports.
+pub fn writer_failed_tasks(writer: &StreamWriter) -> &[i64] {
+    writer
+        .withdrawal
+        .as_ref()
+        .map_or(&[], |w| w.failed_tasks.as_slice())
+}
+
+/// The run's last completed Workflow Task that a writer reports, `None` for none.
+pub fn writer_previous_completed_task(writer: &StreamWriter) -> Option<&WorkflowTaskEpoch> {
+    writer
+        .withdrawal
+        .as_ref()
+        .and_then(|w| w.previous_completed_task.as_ref())
 }
 
 /// The length of a [task_key].
@@ -92,20 +122,24 @@ pub fn task_key(task: &WorkflowTaskEpoch) -> [u8; TASK_KEY_LEN] {
 }
 
 /// The key of an epoch. It is the [task_of] key, then the attempt's scheduled seconds and nanos
-/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and an epoch
+/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and a Producer
 /// with no case is all zeros.
-pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] {
-    let (scheduled, attempt) = match &epoch.kind {
-        Some(Kind::LocalActivity(local)) => (None, local.attempt),
+pub fn epoch_key(kind: Option<&Kind>) -> [u8; EPOCH_KEY_LEN] {
+    let (scheduled, attempt) = match kind {
+        Some(Kind::LocalActivity(local)) => (None, local.epoch.map_or(0, |e| e.attempt)),
         Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => {
-            (activity.attempt_scheduled_time.as_ref(), activity.attempt)
+            let epoch = activity.epoch;
+            (
+                epoch.and_then(|e| e.attempt_scheduled_time),
+                epoch.map_or(0, |e| e.attempt),
+            )
         }
-        Some(Kind::Client(client)) => (None, client.attempt),
+        Some(Kind::Client(client)) => (None, client.epoch.map_or(0, |e| e.attempt)),
         Some(Kind::Workflow(_)) | None => (None, 0),
     };
-    let (seconds, nanos) = time_fields(scheduled);
+    let (seconds, nanos) = time_fields(scheduled.as_ref());
     let mut key = [0; EPOCH_KEY_LEN];
-    key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(epoch)));
+    key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(kind)));
     key[24..32].copy_from_slice(&be64(seconds));
     key[32..36].copy_from_slice(&be32(nanos));
     key[36..].copy_from_slice(&be32(attempt));
@@ -114,16 +148,16 @@ pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] {
 
 /// The Workflow Task of an epoch. It is a Workflow's own task, or the task that dispatched a Local
 /// Activity, and zero for the other kinds.
-pub fn task_of(epoch: &ProducerEpoch) -> WorkflowTaskEpoch {
-    match &epoch.kind {
-        Some(Kind::Workflow(task)) => *task,
-        Some(Kind::LocalActivity(local)) => local.task.unwrap_or_default(),
+pub fn task_of(kind: Option<&Kind>) -> WorkflowTaskEpoch {
+    match kind {
+        Some(Kind::Workflow(w)) => w.epoch.unwrap_or_default(),
+        Some(Kind::LocalActivity(local)) => local.epoch.and_then(|e| e.task).unwrap_or_default(),
         _ => WorkflowTaskEpoch::default(),
     }
 }
 
 /// Orders two epochs field by field.
-pub fn compare_epochs(a: &ProducerEpoch, b: &ProducerEpoch) -> Ordering {
+pub fn compare_epochs(a: Option<&Kind>, b: Option<&Kind>) -> Ordering {
     epoch_key(a).cmp(&epoch_key(b))
 }
 
@@ -150,48 +184,52 @@ pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool {
     let Some(producer) = data.producer.as_ref() else {
         return false;
     };
-    if !writer.withdraw_run.is_empty() && producer.run_id == writer.withdraw_run {
+    let withdraw_run = writer_withdraw_run(writer);
+    if !withdraw_run.is_empty() && producer_run_id(producer) == withdraw_run {
         return true;
     }
     let Some(withdrawing) = writer.producer.as_ref() else {
         return false;
     };
-    let (Some(epoch), Some(old)) = (withdrawing.epoch.as_ref(), producer.epoch.as_ref()) else {
+    let (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref());
+    let (Some(new_kind), Some(old_kind)) = (new, old) else {
         return false;
     };
-    match kind(epoch) {
-        None | Some(ProducerKind::Client) => return false,
-        Some(ProducerKind::Workflow) => {}
+    match ProducerKind::from(new_kind) {
+        ProducerKind::Client => return false,
+        ProducerKind::Workflow => {}
         same => {
-            return kind(old) == same
+            return ProducerKind::from(old_kind) == same
                 && producer.id == withdrawing.id
-                && compare_epochs(old, epoch) == Ordering::Less;
+                && compare_epochs(old, new) == Ordering::Less;
         }
     }
     if !of_run(producer, withdrawing) {
         return false;
     }
-    let (task, old) = (task_of(epoch), task_of(old));
-    let started = old.started_event_id;
-    if writer.failed_tasks.contains(&started) {
+    let (task, old_task) = (task_of(new), task_of(old));
+    let started = old_task.started_event_id;
+    if writer_failed_tasks(writer).contains(&started) {
         return true;
     }
-    let completed = writer.previous_completed_task.as_ref();
+    let completed = writer_previous_completed_task(writer);
     let after = completed.map_or(0, |c| c.started_event_id);
     if after < started && started < task.started_event_id {
         return true;
     }
     let lower_at = |at: &WorkflowTaskEpoch| {
-        started == at.started_event_id && compare_tasks(&old, at) == Ordering::Less
+        started == at.started_event_id && compare_tasks(&old_task, at) == Ordering::Less
     };
     lower_at(&task) || completed.is_some_and(lower_at)
 }
 
 /// Whether a Producer is the Workflow run that withdraws, or one of its Local Activities.
 fn of_run(producer: &Producer, withdrawing: &Producer) -> bool {
-    match producer.epoch.as_ref().and_then(kind) {
+    match kind_of(producer) {
         Some(ProducerKind::Workflow) => producer.id == withdrawing.id,
-        Some(ProducerKind::LocalActivity) => producer.run_id == withdrawing.run_id,
+        Some(ProducerKind::LocalActivity) => {
+            producer_run_id(producer) == producer_run_id(withdrawing)
+        }
         _ => false,
     }
 }
@@ -199,17 +237,14 @@ fn of_run(producer: &Producer, withdrawing: &Producer) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::proto::ClientEpoch;
+    use crate::proto::{ClientEpoch, ClientProducer, producer::Kind};
 
     // Without the clamp, the bytes of a negative field would outrank every real epoch.
     #[test]
     fn a_negative_field_compares_as_zero() {
-        let negative = ProducerEpoch {
-            kind: Some(Kind::Client(ClientEpoch { attempt: -1 })),
-        };
-        assert_eq!(
-            compare_epochs(&negative, &ProducerEpoch::default()),
-            Ordering::Equal
-        );
+        let negative = Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt: -1 }),
+        });
+        assert_eq!(compare_epochs(Some(&negative), None), Ordering::Equal);
     }
 }
diff --git a/crates/streams/src/ids.rs b/crates/streams/src/ids.rs
index d50cf47f3..45eb1e77b 100644
--- a/crates/streams/src/ids.rs
+++ b/crates/streams/src/ids.rs
@@ -11,7 +11,11 @@
 //! script. Each part is percent-encoded, so a `:` or a brace in an id can't make two Streams share
 //! a key.
 
-use crate::{StreamError, StreamResult, epoch::ProducerKind, proto::Producer};
+use crate::{
+    StreamError, StreamResult,
+    epoch::{ProducerKind, producer_run_id},
+    proto::Producer,
+};
 
 /// The longest Stream id, in bytes.
 pub const MAX_STREAM_ID_BYTES: usize = 4096;
@@ -151,13 +155,14 @@ pub fn check_producer(producer: &Producer) -> StreamResult<()> {
         kind,
         ProducerKind::Workflow | ProducerKind::LocalActivity | ProducerKind::Activity
     );
-    if runs && decode_part(parts[0]).as_deref() != Some(producer.run_id.as_str()) {
+    if runs && decode_part(parts[0]).as_deref() != Some(producer_run_id(producer)) {
         return Err(StreamError::invalid(format!(
             "The Producer id {:?} names another run than its run id {:?}",
-            producer.id, producer.run_id
+            producer.id,
+            producer_run_id(producer)
         )));
     }
-    if kind == ProducerKind::Client && !producer.run_id.is_empty() {
+    if kind == ProducerKind::Client && !producer_run_id(producer).is_empty() {
         return Err(StreamError::invalid("A client Producer has no run id"));
     }
     Ok(())
@@ -207,20 +212,34 @@ impl StreamKeys {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::proto::{ProducerEpoch, producer_epoch::Kind};
+    use crate::proto::{
+        ActivityProducer, ClientProducer, LocalActivityProducer, WorkflowProducer, producer::Kind,
+    };
 
     fn producer(kind: ProducerKind, id: &str, run_id: &str) -> Producer {
+        let run_id = run_id.to_string();
         let kind = match kind {
-            ProducerKind::Workflow => Kind::Workflow(Default::default()),
-            ProducerKind::LocalActivity => Kind::LocalActivity(Default::default()),
-            ProducerKind::Activity => Kind::Activity(Default::default()),
-            ProducerKind::StandaloneActivity => Kind::StandaloneActivity(Default::default()),
-            ProducerKind::Client => Kind::Client(Default::default()),
+            ProducerKind::Workflow => Kind::Workflow(WorkflowProducer {
+                run_id,
+                ..Default::default()
+            }),
+            ProducerKind::LocalActivity => Kind::LocalActivity(LocalActivityProducer {
+                run_id,
+                ..Default::default()
+            }),
+            ProducerKind::Activity => Kind::Activity(ActivityProducer {
+                run_id,
+                ..Default::default()
+            }),
+            ProducerKind::StandaloneActivity => Kind::StandaloneActivity(ActivityProducer {
+                run_id,
+                ..Default::default()
+            }),
+            ProducerKind::Client => Kind::Client(ClientProducer::default()),
         };
         Producer {
             id: id.to_string(),
-            epoch: Some(ProducerEpoch { kind: Some(kind) }),
-            run_id: run_id.to_string(),
+            kind: Some(kind),
         }
     }
 
@@ -267,14 +286,9 @@ mod tests {
             producer(ProducerKind::Workflow, &workflow_producer_id(run), "other"),
             producer(ProducerKind::Activity, "act:run%3A1:a", run),
             Producer {
-                epoch: None,
-                ..producer(ProducerKind::Workflow, "wf:x", "x")
-            },
-            Producer {
-                epoch: Some(ProducerEpoch::default()),
+                kind: None,
                 ..producer(ProducerKind::Workflow, "wf:x", "x")
             },
-            producer(ProducerKind::Client, "client:a:b", "run"),
             producer(ProducerKind::Client, "client:a|b", ""),
             producer(ProducerKind::Client, "client:run%3a1", ""),
         ];
diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs
index 2bd2cba35..8c87e1fbf 100644
--- a/crates/streams/src/lib.rs
+++ b/crates/streams/src/lib.rs
@@ -10,6 +10,7 @@ pub mod digest;
 pub mod epoch;
 mod error;
 pub mod ids;
+pub mod memory;
 pub mod offset;
 pub mod store;
 #[cfg(any(test, feature = "test-support"))]
diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs
new file mode 100644
index 000000000..4cd315d1f
--- /dev/null
+++ b/crates/streams/src/memory.rs
@@ -0,0 +1,898 @@
+//! The memory Store, for tests. It keeps the same fields as the Redis layout and runs the steps of
+//! the Lua scripts in the same order, so that both Stores return the same results. Each call holds
+//! one lock, since Redis runs one script at a time.
+
+use crate::{
+    Reason, StreamError, StreamResult,
+    digest::DIGEST_LEN,
+    epoch::{EPOCH_KEY_LEN, ProducerKind, epoch_key},
+    ids::StreamKeys,
+    offset::{entry_id_of_offset, offset_of_entry_id},
+    proto::{Producer, StreamRef},
+    store::{
+        Appended, Batch, Capabilities, Closed, ConflictPolicy, Create, CreateOutcome, Created,
+        Entry, FORMAT_VERSION, MAX_SEQUENCE, Owner, Store, StreamState, TOMBSTONE, owner_chain,
+    },
+};
+use parking_lot::Mutex;
+use std::{
+    collections::{BTreeMap, HashMap},
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use tokio::{sync::Notify, time::Instant};
+
+/// The number of sequence numbers in one millisecond, since an offset packs them in 20 bits.
+const SEQ_LIMIT: u64 = 1 << crate::offset::SEQ_BITS;
+/// The most Producers that one append prunes, as `PRUNE_LIMIT` in `lib.lua`.
+const PRUNE_LIMIT: usize = 64;
+
+#[derive(Clone, Default)]
+struct Fields {
+    record: Option<Vec<u8>>,
+    superseded: Option<Vec<u8>>,
+}
+
+/// One Stream's meta hash and log, each with an expiry time, as Redis keys have.
+#[derive(Default)]
+struct Keys {
+    /// The meta hash, whose field names and values are bytes as in Redis.
+    meta: HashMap<Vec<u8>, Vec<u8>>,
+    meta_expires: i64,
+    /// `None` when the log key doesn't exist.
+    log: Option<BTreeMap<(u64, u64), Fields>>,
+    log_expires: i64,
+    /// The newest id of the log, which Redis keeps after a trim and drops with the expired key.
+    last_id: (u64, u64),
+    /// The `producers` sorted set, each `<kind>|<producer id>` with the ms of its last append.
+    producers: HashMap<String, i64>,
+    producers_expires: i64,
+}
+
+/// A Store in process memory, with the rules of the Redis scripts.
+pub struct MemoryStore {
+    prefix: String,
+    retention: Duration,
+    safety_ttl: Duration,
+    streams: Mutex<HashMap<String, Keys>>,
+    appended: Notify,
+}
+
+impl std::fmt::Debug for MemoryStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemoryStore")
+            .field("retention", &self.retention)
+            .field("safety_ttl", &self.safety_ttl)
+            .finish_non_exhaustive()
+    }
+}
+
+fn now_ms() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |since| since.as_millis() as i64)
+}
+
+fn millis(duration: Duration) -> i64 {
+    i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
+}
+
+fn refuse(reason: Reason, message: &str) -> StreamError {
+    StreamError::new(reason, message)
+}
+
+fn id_text((ms, seq): (u64, u64)) -> String {
+    format!("{ms}-{seq}")
+}
+
+fn parse_id(text: &str) -> Option<(u64, u64)> {
+    let (ms, seq) = text.split_once('-')?;
+    Some((ms.parse().ok()?, seq.parse().ok()?))
+}
+
+/// The length of a [ProducerField] value.
+const PRODUCER_FIELD_LEN: usize = 1 + 2 * EPOCH_KEY_LEN + 8 + 2 * 12 + DIGEST_LEN;
+/// The flag of a [ProducerField] that has a held epoch. The other flag bits are zero.
+const HELD: u8 = 1;
+
+/// An entry id as a [ProducerField] keeps it, the ms and the seq.
+pub(crate) type EntryId = (u64, u32);
+
+/// The `p|<kind>|<producer id>` field, one per Producer. Its value is 129 bytes: a flag byte, the
+/// fence, the held epoch key or zeros, and the last batch's sequence (8), first and last entry ids
+/// (12 each, the ms in 8 and the seq in 4, zeros for none) and digest. Numbers are big-endian.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct ProducerField {
+    /// The newest epoch key of the Producer, which is also the epoch of its last batch.
+    pub(crate) fence: [u8; EPOCH_KEY_LEN],
+    /// The newest epoch key that still has Records.
+    pub(crate) held: Option<[u8; EPOCH_KEY_LEN]>,
+    /// The last batch's sequence.
+    pub(crate) sequence: i64,
+    /// The last batch's first and last entry ids, `None` when it wrote no entry.
+    pub(crate) entries: Option<(EntryId, EntryId)>,
+    /// The last batch's digest.
+    pub(crate) digest: [u8; DIGEST_LEN],
+}
+
+impl ProducerField {
+    pub(crate) fn encode(&self) -> Vec<u8> {
+        let mut value = Vec::with_capacity(PRODUCER_FIELD_LEN);
+        value.push(if self.held.is_some() { HELD } else { 0 });
+        value.extend_from_slice(&self.fence);
+        value.extend_from_slice(&self.held.unwrap_or([0; EPOCH_KEY_LEN]));
+        value.extend_from_slice(&self.sequence.to_be_bytes());
+        let (first, last) = self.entries.unwrap_or_default();
+        for (ms, seq) in [first, last] {
+            value.extend_from_slice(&ms.to_be_bytes());
+            value.extend_from_slice(&seq.to_be_bytes());
+        }
+        value.extend_from_slice(&self.digest);
+        value
+    }
+}
+
+/// Refuses a value of another length or with another flag bit, which a later format may write.
+impl TryFrom<&[u8]> for ProducerField {
+    type Error = StreamError;
+
+    fn try_from(value: &[u8]) -> StreamResult<Self> {
+        let unknown = || {
+            refuse(
+                Reason::StorageCorrupt,
+                "the Producer field has an unknown format",
+            )
+        };
+        let Some((&flags, rest)) = value.split_first() else {
+            return Err(unknown());
+        };
+        if value.len() != PRODUCER_FIELD_LEN || flags & !HELD != 0 {
+            return Err(unknown());
+        }
+        let (fence, rest) = rest.split_at(EPOCH_KEY_LEN);
+        let (held, rest) = rest.split_at(EPOCH_KEY_LEN);
+        let (sequence, rest) = rest.split_at(8);
+        let (first, rest) = rest.split_at(12);
+        let (last, digest) = rest.split_at(12);
+        let entry_id = |bytes: &[u8]| -> StreamResult<EntryId> {
+            let (ms, seq) = bytes.split_at(8);
+            let ms = ms.try_into().map_err(|_| unknown())?;
+            let seq = seq.try_into().map_err(|_| unknown())?;
+            Ok((u64::from_be_bytes(ms), u32::from_be_bytes(seq)))
+        };
+        // A batch has both entry ids or neither.
+        let entries = match (entry_id(first)?, entry_id(last)?) {
+            ((0, 0), (0, 0)) => None,
+            ((0, 0), _) | (_, (0, 0)) => return Err(unknown()),
+            ids => Some(ids),
+        };
+        Ok(Self {
+            fence: fence.try_into().map_err(|_| unknown())?,
+            held: if flags & HELD != 0 {
+                Some(held.try_into().map_err(|_| unknown())?)
+            } else {
+                None
+            },
+            sequence: i64::from_be_bytes(sequence.try_into().map_err(|_| unknown())?),
+            entries,
+            digest: digest.try_into().map_err(|_| unknown())?,
+        })
+    }
+}
+
+/// The [EntryId] of an entry id's text, `<ms>-<seq>`.
+fn entry_id(text: &str) -> StreamResult<EntryId> {
+    let bad = || refuse(Reason::StorageCorrupt, "an entry id isn't <ms>-<seq>");
+    let (ms, seq) = text.split_once('-').ok_or_else(bad)?;
+    Ok((
+        ms.parse().map_err(|_| bad())?,
+        seq.parse().map_err(|_| bad())?,
+    ))
+}
+
+impl Keys {
+    fn get(&self, field: impl AsRef<[u8]>) -> Option<&[u8]> {
+        self.meta.get(field.as_ref()).map(Vec::as_slice)
+    }
+
+    /// A field that holds text, such as `own` or `newest`. `None` when it's absent or not text.
+    fn text(&self, field: impl AsRef<[u8]>) -> Option<&str> {
+        self.get(field)
+            .and_then(|value| std::str::from_utf8(value).ok())
+    }
+
+    /// A field that holds a number, zero when it's absent.
+    fn number(&self, field: &str) -> i64 {
+        self.text(field)
+            .and_then(|number| number.parse().ok())
+            .unwrap_or(0)
+    }
+
+    fn set(&mut self, field: impl Into<Vec<u8>>, value: impl Into<Vec<u8>>) {
+        self.meta.insert(field.into(), value.into());
+    }
+
+    fn remove(&mut self, field: impl AsRef<[u8]>) {
+        self.meta.remove(field.as_ref());
+    }
+
+    /// The `check_live` step, which checks the create key, the format version and the expiry.
+    fn check_live(&self, now: i64) -> StreamResult<()> {
+        if self.get("v").is_none() {
+            return Err(refuse(Reason::NotFound, "nobody created the Stream"));
+        }
+        let v = self.number("v");
+        if v > i64::from(FORMAT_VERSION) {
+            return Err(refuse(
+                Reason::FormatTooNew,
+                &format!("format {v} is newer than {FORMAT_VERSION}"),
+            ));
+        }
+        if self
+            .text("exp")
+            .and_then(|exp| exp.parse::<i64>().ok())
+            .is_some_and(|exp| exp <= now)
+        {
+            return Err(refuse(Reason::Expired, "the Stream expired"));
+        }
+        Ok(())
+    }
+
+    /// The `check_owner` step of a close, as in `lib.lua`.
+    fn check_owner(&self, kind: ProducerKind, chain: &str) -> StreamResult<()> {
+        if self.text("own") == Some("wf") {
+            if kind != ProducerKind::Workflow {
+                return Err(refuse(
+                    Reason::NotOwner,
+                    "only the close Activity of the owner run chain closes a Workflow-owned Stream",
+                ));
+            }
+            if self.text("oc").is_some_and(|owner| chain != owner) {
+                return Err(refuse(
+                    Reason::NotOwner,
+                    "the ref has another run chain than the owner",
+                ));
+            }
+        } else if !matches!(kind, ProducerKind::Workflow | ProducerKind::Client) {
+            return Err(refuse(
+                Reason::NotOwner,
+                "only a Workflow or a client closes a Stream with no owner",
+            ));
+        }
+        Ok(())
+    }
+
+    fn refresh(&mut self, now: i64, safety: i64, tombstone: i64) {
+        let retention = self.number("ret");
+        let mut ttl = retention;
+        if self.text("own") == Some("wf") && self.get("closed").is_none() {
+            ttl = safety;
+        }
+        let exp = now.saturating_add(ttl);
+        self.set("exp", exp.to_string());
+        self.meta_expires = exp.saturating_add(tombstone);
+        if self.log.is_some() {
+            // A read hides every Record one Retention after the last write, so the log goes then.
+            self.log_expires = exp.min(now.saturating_add(retention));
+        }
+        if !self.producers.is_empty() {
+            self.producers_expires = exp;
+        }
+    }
+
+    /// Removes the `p|` field of the Producers whose last append is older than the Retention, as
+    /// `prune` in `lib.lua`. Their Records are older too, so they hold none.
+    fn prune(&mut self, now: i64) {
+        let retention = self.number("ret");
+        let min = now - retention;
+        let mut old: Vec<_> = self
+            .producers
+            .iter()
+            .filter(|(_, at)| **at < min)
+            .map(|(producer, at)| (*at, producer.clone()))
+            .collect();
+        old.sort();
+        for (_, producer) in old.into_iter().take(PRUNE_LIMIT) {
+            self.remove(format!("p|{producer}"));
+            self.producers.remove(&producer);
+        }
+    }
+
+    /// The newest id the Stream gave. Redis drops it with an expired log, so the meta's `newest`
+    /// keeps the next ids above the old ones.
+    fn newest_id(&self) -> (u64, u64) {
+        let newest = self.text("newest").and_then(parse_id);
+        self.last_id.max(newest.unwrap_or_default())
+    }
+
+    /// The ms of the ids that the next `add` gives, as `next_ms` in `lib.lua`.
+    fn next_ms(&self, now: i64) -> i64 {
+        now.max(self.newest_id().0 as i64)
+    }
+
+    /// Refuses a create that names another owner than the Stream has, as `check_identity` in
+    /// `lib.lua`.
+    fn check_identity(&self, owner: &str, chain: &str) -> StreamResult<()> {
+        let stored = self.text("oc").unwrap_or_default();
+        if self.text("own") != Some(owner) || stored != chain {
+            return Err(refuse(
+                Reason::NotOwner,
+                "the Stream has another owner than the create names",
+            ));
+        }
+        Ok(())
+    }
+
+    fn trim(&mut self, now: i64) {
+        if self.log.is_none() {
+            // The expired log took every Record, so a read from an older offset must fail.
+            if let Some(newest) = self.text("newest").and_then(parse_id) {
+                self.raise_trimmed(newest);
+            }
+            return;
+        }
+        let retention = self.number("ret");
+        let min = now - retention;
+        if min <= 0 {
+            return;
+        }
+        let Some(log) = self.log.as_ref() else {
+            return;
+        };
+        let old: Vec<_> = log.range(..(min as u64, 0)).map(|(id, _)| *id).collect();
+        let Some(newest_old) = old.last().copied() else {
+            return;
+        };
+        self.raise_trimmed(newest_old);
+        if let Some(log) = self.log.as_mut() {
+            for id in old {
+                log.remove(&id);
+            }
+        }
+    }
+
+    /// Sets the trimmed offset to `id`, unless it is already at or past it.
+    fn raise_trimmed(&mut self, id: (u64, u64)) {
+        let moves = self
+            .text("trimmed")
+            .and_then(parse_id)
+            .is_none_or(|trimmed| trimmed < id);
+        if moves {
+            self.set("trimmed", id_text(id));
+        }
+    }
+
+    /// Refuses `count` entries whose ids would not fit an offset. The ids go on from the newest id
+    /// the Stream gave, as `check_ids_fit` in `lib.lua`.
+    fn check_ids_fit(&self, now: i64, count: u64) -> StreamResult<()> {
+        let (ms, seq) = self.newest_id();
+        let first = if now > ms as i64 { 0 } else { seq + 1 };
+        if first + count > SEQ_LIMIT {
+            return Err(refuse(
+                Reason::StorageUnavailable,
+                "the entry ids of this millisecond would not fit an offset",
+            ));
+        }
+        Ok(())
+    }
+
+    /// Adds an entry as `add` in `lib.lua` does, with the next id after the newest one, in this
+    /// millisecond or a later one.
+    fn add(&mut self, now: i64, fields: Fields) -> String {
+        let now = now.max(0) as u64;
+        let (last_ms, last_seq) = self.newest_id();
+        let id = if now > last_ms {
+            (now, 0)
+        } else {
+            (last_ms, last_seq + 1)
+        };
+        self.last_id = id;
+        self.log
+            .get_or_insert_with(BTreeMap::new)
+            .insert(id, fields);
+        id_text(id)
+    }
+}
+
+impl MemoryStore {
+    /// An empty Store.
+    pub fn new(retention: Duration, safety_ttl: Duration) -> Self {
+        Self {
+            prefix: crate::ids::DEFAULT_KEY_PREFIX.to_string(),
+            retention,
+            safety_ttl,
+            streams: Mutex::default(),
+            appended: Notify::new(),
+        }
+    }
+
+    fn key(&self, namespace: &str, stream_id: &str) -> String {
+        StreamKeys::new(&self.prefix, namespace, stream_id).meta()
+    }
+
+    /// Runs `step` on one Stream's keys after it drops the expired keys, as Redis does.
+    fn with<T>(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        step: impl FnOnce(&mut Keys, i64) -> T,
+    ) -> T {
+        let now = now_ms();
+        let mut streams = self.streams.lock();
+        let key = self.key(namespace, stream_id);
+        let keys = streams.entry(key.clone()).or_default();
+        if keys.meta_expires > 0 && keys.meta_expires <= now {
+            keys.meta.clear();
+            keys.meta_expires = 0;
+        }
+        if keys.log_expires > 0 && keys.log_expires <= now {
+            keys.log = None;
+            keys.log_expires = 0;
+            keys.last_id = (0, 0);
+        }
+        if keys.producers_expires > 0 && keys.producers_expires <= now {
+            keys.producers.clear();
+            keys.producers_expires = 0;
+        }
+        let result = step(keys, now);
+        // Once every key is gone, Redis has no state of the Stream, not even its newest id.
+        if keys.meta.is_empty() && keys.log.is_none() && keys.producers.is_empty() {
+            streams.remove(&key);
+        }
+        result
+    }
+}
+
+#[cfg(any(test, feature = "test-support"))]
+impl MemoryStore {
+    /// Changes one meta field, for tests of what the scripts refuse.
+    #[doc(hidden)]
+    pub fn test_set_meta(&self, namespace: &str, stream_id: &str, field: &[u8], value: &[u8]) {
+        self.with(namespace, stream_id, |keys, _| keys.set(field, value));
+    }
+
+    /// One meta field, for tests.
+    #[doc(hidden)]
+    pub fn test_meta(&self, namespace: &str, stream_id: &str, field: &[u8]) -> Option<Vec<u8>> {
+        self.with(namespace, stream_id, |keys, _| {
+            keys.get(field).map(<[u8]>::to_vec)
+        })
+    }
+
+    /// The ms left on the log and on the meta, -2 for a key that doesn't exist, as `PTTL`.
+    #[doc(hidden)]
+    pub fn test_ttls(&self, namespace: &str, stream_id: &str) -> (i64, i64) {
+        self.with(namespace, stream_id, |keys, now| {
+            let log = if keys.log.is_some() {
+                keys.log_expires - now
+            } else {
+                -2
+            };
+            let meta = if keys.meta.is_empty() {
+                -2
+            } else {
+                keys.meta_expires - now
+            };
+            (log, meta)
+        })
+    }
+
+    /// Adds an entry with an explicit id, as `XADD <id>`, for tests.
+    #[doc(hidden)]
+    pub fn test_add_entry(&self, namespace: &str, stream_id: &str, id: &str) {
+        self.with(namespace, stream_id, |keys, _| {
+            let id = parse_id(id).expect("an entry id");
+            keys.last_id = keys.last_id.max(id);
+            keys.log
+                .get_or_insert_with(BTreeMap::new)
+                .insert(id, Fields::default());
+        });
+    }
+
+    /// The number of log entries, as `XLEN`, for tests.
+    #[doc(hidden)]
+    pub fn test_log_len(&self, namespace: &str, stream_id: &str) -> usize {
+        self.with(namespace, stream_id, |keys, _| {
+            keys.log.as_ref().map_or(0, BTreeMap::len)
+        })
+    }
+}
+
+fn offsets(first: &str, last: &str) -> StreamResult<Option<(i64, i64)>> {
+    if first.is_empty() || last.is_empty() {
+        return Ok(None);
+    }
+    Ok(Some((
+        offset_of_entry_id(first)?,
+        offset_of_entry_id(last)?,
+    )))
+}
+
+#[async_trait::async_trait]
+impl Store for MemoryStore {
+    fn capabilities(&self) -> Capabilities {
+        Capabilities::ALL.without(Capabilities::RUN_FENCE)
+    }
+
+    async fn create(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        create: &Create<'_>,
+    ) -> StreamResult<Created> {
+        let stream_id = stream_ref.id.as_str();
+        let retention = create.retention.filter(|r| !r.is_zero());
+        let retention = millis(retention.unwrap_or(self.retention));
+        let request_id = create.request_id;
+        let owner = match create.owner {
+            Owner::Workflow => "wf",
+            Owner::Client => "client",
+        };
+        let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE));
+        let chain = if owner == "wf" {
+            owner_chain(stream_ref)
+        } else {
+            String::new()
+        };
+        let created = self.with(namespace, stream_id, |keys, now| {
+            if keys.get("ret").is_none() {
+                keys.set("v", FORMAT_VERSION.to_string());
+                keys.set("ret", retention.to_string());
+                keys.set("own", owner);
+                if !chain.is_empty() {
+                    keys.set("oc", chain.clone());
+                }
+                if !request_id.is_empty() {
+                    keys.set("rid", request_id);
+                }
+                keys.refresh(now, safety, tombstone);
+                return Ok((CreateOutcome::Created, retention));
+            }
+            keys.check_live(now)?;
+            keys.check_identity(owner, &chain)?;
+            if keys.get("closed").is_some() {
+                return Err(refuse(Reason::Closed, "the Stream is closed"));
+            }
+            // A retry of the create that made the Stream gets its answer again.
+            if !request_id.is_empty() && keys.get("rid") == Some(request_id.as_bytes()) {
+                return Ok((CreateOutcome::Created, keys.number("ret")));
+            }
+            if create.conflict == ConflictPolicy::Fail {
+                return Err(refuse(Reason::AlreadyExists, "the Stream exists"));
+            }
+            Ok((CreateOutcome::Exists, keys.number("ret")))
+        });
+        let (outcome, stored) = created?;
+        Ok(Created {
+            outcome,
+            retention: Duration::from_millis(stored.max(0) as u64),
+            flagged: false,
+        })
+    }
+
+    async fn append(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        batch: &Batch<'_>,
+    ) -> StreamResult<Appended> {
+        let stream_id = stream_ref.id.as_str();
+        let producer = batch.producer;
+        let kind = ProducerKind::try_from(producer)?;
+        let epoch = epoch_key(producer.kind.as_ref());
+        let withdrawal = &batch.withdrawal;
+        let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE));
+        let field = format!("p|{}|{}", kind as i32, producer.id);
+        let member = format!("{}|{}", kind as i32, producer.id);
+        let appended = self.with(namespace, stream_id, |keys, now| {
+            // The format version comes first, since a newer format can change the Producer field.
+            keys.check_live(now)?;
+            let stored = keys.get(&field).map(ProducerField::try_from).transpose()?;
+            // Every append sets the fence to its epoch, so the fence is the last batch's epoch.
+            let previous = stored.filter(|stored| stored.fence == epoch);
+            // The lock also covers the dedupe check, so no other batch takes the same number.
+            let sequence = match batch.sequence {
+                0 => previous.map_or(1, |previous| previous.sequence + 1),
+                sequence => sequence,
+            };
+            if !(1..=MAX_SEQUENCE).contains(&sequence) {
+                return Err(refuse(
+                    Reason::Invalid,
+                    "the sequence must be from 1 to 2^53 - 1",
+                ));
+            }
+            if let Some(previous) = previous {
+                if sequence == previous.sequence {
+                    if previous.digest != batch.digest {
+                        return Err(refuse(
+                            Reason::SeqDivergent,
+                            &format!(
+                                "sequence {sequence} has other content than the batch the Store \
+                                 has"
+                            ),
+                        ));
+                    }
+                    let (first, last) = previous
+                        .entries
+                        .map(|((ms, seq), (last_ms, last_seq))| {
+                            (format!("{ms}-{seq}"), format!("{last_ms}-{last_seq}"))
+                        })
+                        .unwrap_or_default();
+                    return Ok((true, first, last, false));
+                }
+                if sequence < previous.sequence {
+                    return Err(refuse(
+                        Reason::SequenceTooLow,
+                        &format!(
+                            "sequence {sequence} is below the last one, {}",
+                            previous.sequence
+                        ),
+                    ));
+                }
+            }
+            if keys.get("closed").is_some() {
+                return Err(refuse(Reason::Closed, "the Stream is closed"));
+            }
+            if stored.is_some_and(|stored| epoch < stored.fence) {
+                return Err(refuse(
+                    Reason::Fenced,
+                    "the epoch is lower than the newest one of the Producer",
+                ));
+            }
+            let records = batch.records.len() as u64;
+            keys.check_ids_fit(now, records + 1)?;
+
+            // A client's newer epoch only fences, as in `append.lua`.
+            let flagged = kind != ProducerKind::Client
+                && stored
+                    .and_then(|stored| stored.held)
+                    .is_some_and(|held| held < epoch);
+
+            keys.trim(now);
+            keys.prune(now);
+            let at = keys.next_ms(now);
+            let (mut first, mut last) = (String::new(), String::new());
+            if records == 0 && flagged {
+                last = keys.add(
+                    now,
+                    Fields {
+                        record: None,
+                        superseded: Some(withdrawal.superseded.to_vec()),
+                    },
+                );
+                first = last.clone();
+            }
+            for (i, record) in batch.records.iter().enumerate() {
+                let superseded = (i == 0 && flagged).then(|| withdrawal.superseded.to_vec());
+                last = keys.add(
+                    now,
+                    Fields {
+                        record: Some(record.clone()),
+                        superseded,
+                    },
+                );
+                if first.is_empty() {
+                    first = last.clone();
+                }
+            }
+
+            let held = if records > 0 {
+                Some(epoch)
+            } else if flagged {
+                None
+            } else {
+                stored.and_then(|stored| stored.held)
+            };
+            let entries = if last.is_empty() {
+                None
+            } else {
+                Some((entry_id(&first)?, entry_id(&last)?))
+            };
+            let value = ProducerField {
+                fence: epoch,
+                held,
+                sequence,
+                entries,
+                digest: batch.digest,
+            };
+            keys.set(field.as_str(), value.encode());
+            if !last.is_empty() {
+                keys.set("newest", last.clone());
+            }
+            keys.producers.insert(member.clone(), at);
+            keys.refresh(now, safety, tombstone);
+            Ok((false, first, last, flagged))
+        });
+        let (resend, first, last, flagged) = appended?;
+        if !resend && !last.is_empty() {
+            self.appended.notify_waiters();
+        }
+        Ok(Appended {
+            offsets: offsets(&first, &last)?,
+            resend,
+            flagged,
+        })
+    }
+
+    async fn close(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        producer: &Producer,
+    ) -> StreamResult<Closed> {
+        let stream_id = stream_ref.id.as_str();
+        let kind = ProducerKind::try_from(producer)?;
+        let epoch = epoch_key(producer.kind.as_ref());
+        let field = format!("p|{}|{}", kind as i32, producer.id);
+        let tombstone = millis(TOMBSTONE);
+        let chain = owner_chain(stream_ref);
+        let closed = self.with(namespace, stream_id, |keys, now| {
+            keys.check_live(now)?;
+            keys.check_owner(kind, &chain)?;
+            if keys.get("closed").is_some() {
+                return Ok(true);
+            }
+            let stored = keys.get(&field).map(ProducerField::try_from).transpose()?;
+            if stored.is_some_and(|stored| epoch < stored.fence) {
+                return Err(refuse(
+                    Reason::Fenced,
+                    "the epoch is lower than the newest one of the Producer",
+                ));
+            }
+            keys.set("closed", "1");
+            keys.refresh(now, 0, tombstone);
+            Ok(false)
+        });
+        let already = closed?;
+        // No entry wakes a waiting read, so the close wakes it.
+        if !already {
+            self.appended.notify_waiters();
+        }
+        Ok(Closed { already })
+    }
+
+    async fn read(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        from: i64,
+        count: usize,
+    ) -> StreamResult<(StreamState, Vec<Entry>)> {
+        let start = if from > 0 {
+            parse_id(&entry_id_of_offset(from)).unwrap_or((0, 0))
+        } else {
+            (0, 0)
+        };
+        self.with(namespace, stream_id, |keys, now| {
+            let offset = |field: &str| -> StreamResult<i64> {
+                keys.text(field).map_or(Ok(0), offset_of_entry_id)
+            };
+            let number = |field: &str| keys.number(field);
+            let state = StreamState {
+                exists: keys.get("v").is_some(),
+                version: number("v") as u32,
+                closed: keys.get("closed").is_some(),
+                newest: offset("newest")?,
+                trimmed: offset("trimmed")?,
+                retention: Duration::from_millis(number("ret").max(0) as u64),
+                expires_ms: number("exp"),
+                now_ms: now,
+            };
+            let mut entries = vec![];
+            if let Some(log) = keys.log.as_ref() {
+                for (id, fields) in log.range(start..).take(count) {
+                    entries.push(Entry {
+                        offset: offset_of_entry_id(&id_text(*id))?,
+                        record: fields.record.clone(),
+                        superseded: fields.superseded.clone(),
+                    });
+                }
+            }
+            Ok((state, entries))
+        })
+    }
+
+    async fn read_back(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        before: i64,
+        count: usize,
+    ) -> StreamResult<Vec<Entry>> {
+        let before = parse_id(&entry_id_of_offset(before)).unwrap_or((0, 0));
+        self.with(namespace, stream_id, |keys, _| {
+            let Some(log) = keys.log.as_ref() else {
+                return Ok(vec![]);
+            };
+            log.range(..=before)
+                .rev()
+                .take(count)
+                .map(|(id, fields)| {
+                    Ok(Entry {
+                        offset: offset_of_entry_id(&id_text(*id))?,
+                        record: fields.record.clone(),
+                        superseded: fields.superseded.clone(),
+                    })
+                })
+                .collect()
+        })
+    }
+
+    async fn wait_after(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        after: i64,
+        deadline: Instant,
+    ) -> StreamResult<bool> {
+        let after = if after > 0 {
+            parse_id(&entry_id_of_offset(after)).unwrap_or((0, 0))
+        } else {
+            (0, 0)
+        };
+        loop {
+            // Listen before the check, so an append between the two still wakes the wait.
+            let appended = self.appended.notified();
+            let found = self.with(namespace, stream_id, |keys, _| {
+                keys.get("closed").is_some()
+                    || keys
+                        .log
+                        .as_ref()
+                        .and_then(|log| log.keys().next_back())
+                        .is_some_and(|id| *id > after)
+            });
+            if found {
+                return Ok(true);
+            }
+            if tokio::time::timeout_at(deadline, appended).await.is_err() {
+                return Ok(false);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn an_id_follows_the_newest_one_within_a_millisecond() {
+        let mut keys = Keys::default();
+        assert_eq!(keys.add(5, Fields::default()), "5-0");
+        assert_eq!(keys.add(5, Fields::default()), "5-1");
+        // The ids keep increasing when the clock steps back.
+        assert_eq!(keys.add(3, Fields::default()), "5-2");
+        assert_eq!(keys.add(9, Fields::default()), "9-0");
+    }
+
+    #[test]
+    fn a_producer_field_matches_the_shared_vectors() {
+        let vectors: serde_json::Value =
+            serde_json::from_str(include_str!("../testdata/fields.json")).unwrap();
+        let hex = |text: &str| -> Vec<u8> {
+            (0..text.len())
+                .step_by(2)
+                .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap())
+                .collect()
+        };
+        for case in vectors["producer_values"].as_array().unwrap() {
+            let text = |name: &str| case[name].as_str().unwrap();
+            let key = |name: &str| case[name].as_str().map(|key| hex(key).try_into().unwrap());
+            let entries = (!text("first").is_empty()).then(|| {
+                (
+                    entry_id(text("first")).unwrap(),
+                    entry_id(text("last")).unwrap(),
+                )
+            });
+            let field = ProducerField {
+                fence: key("fence").unwrap(),
+                held: key("held"),
+                sequence: case["sequence"].as_i64().unwrap(),
+                entries,
+                digest: hex(text("digest")).try_into().unwrap(),
+            };
+            let value = hex(text("value"));
+            assert_eq!(field.encode(), value, "{}", text("name"));
+            assert_eq!(ProducerField::try_from(value.as_slice()).unwrap(), field);
+        }
+    }
+}
diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs
index 97bac1055..7a0e8d1ac 100644
--- a/crates/streams/src/store.rs
+++ b/crates/streams/src/store.rs
@@ -11,6 +11,13 @@ use std::time::Duration;
 use temporalio_common::protos::temporal::api::enums::v1::ExecutionType;
 use tokio::time::Instant;
 
+/// The Retention of a Stream whose create sets none, when a Store's options set no other.
+pub const DEFAULT_RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60);
+
+/// How long an open Workflow-owned Stream stays after its last write, when a Store's options set
+/// no other.
+pub const DEFAULT_SAFETY_TTL: Duration = Duration::from_secs(90 * 24 * 60 * 60);
+
 /// The format version this crate writes. A Stream with a higher one refuses writes.
 pub const FORMAT_VERSION: u32 = 1;
 /// The maximum size of a stored Record, in bytes.
diff --git a/crates/streams/src/testing/fields.rs b/crates/streams/src/testing/fields.rs
index f05c80d6d..5d1620166 100644
--- a/crates/streams/src/testing/fields.rs
+++ b/crates/streams/src/testing/fields.rs
@@ -6,21 +6,19 @@ use crate::ids::encode_part;
 /// The format version of the Stream.
 pub const VERSION: &str = "v";
 
-/// The newest epoch key of a Producer, the fence.
-pub fn fence(kind: i32, producer_id: &str) -> String {
-    format!("e|{kind}|{producer_id}")
-}
-
-/// The last batch of a Producer's newest epoch, as
-/// `<epoch key>|<sequence>|<first>|<last>|<digest>`. These are the dedupe key, the first and last
-/// entry ids of the batch, and the content digest that a resend must match.
-pub fn dedupe(kind: i32, producer_id: &str) -> String {
-    format!("d|{kind}|{producer_id}")
-}
-
-/// The newest epoch key of a Producer that still has Records in the Stream.
-pub fn holds(kind: i32, producer_id: &str) -> String {
-    format!("h|{kind}|{producer_id}")
+/// The field of a Producer. Its value is 129 bytes, with big-endian numbers:
+///
+/// - 1, the flags. Bit 0 is set when the held epoch is present, and the other bits are zero.
+/// - 40, the fence, the Producer's newest epoch key. It is also the epoch of the last batch.
+/// - 40, the held epoch key, the newest one that still has Records, or zeros.
+/// - 8, the last batch's sequence.
+/// - 12 and 12, the last batch's first and last entry ids, each the ms in 8 bytes and the seq in
+///   4. They are zero when the batch wrote no entry.
+/// - 16, the last batch's digest, which a resend must match.
+///
+/// A Store refuses a value of another length or with another flag bit.
+pub fn producer(kind: i32, producer_id: &str) -> String {
+    format!("p|{kind}|{producer_id}")
 }
 
 /// Set when a Producer of the run wrote a Record to the Stream.
@@ -43,10 +41,8 @@ mod tests {
             serde_json::from_str(include_str!("../../testdata/fields.json")).unwrap();
         let text = |field: &str| case[field].as_str().unwrap();
         let kind = case["kind"].as_i64().unwrap() as i32;
-        let (producer, run) = (text("producer_id"), text("run_id"));
-        assert_eq!(fence(kind, producer), text("fence"));
-        assert_eq!(dedupe(kind, producer), text("dedupe"));
-        assert_eq!(holds(kind, producer), text("holds"));
+        let (producer_id, run) = (text("producer_id"), text("run_id"));
+        assert_eq!(producer(kind, producer_id), text("producer"));
         assert_eq!(run_wrote(run), text("run_wrote"));
         assert_eq!(withdrawn_run(run), text("withdrawn_run"));
     }
diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs
index f8feafa3b..0b2aa498d 100644
--- a/crates/streams/src/testing/mod.rs
+++ b/crates/streams/src/testing/mod.rs
@@ -1,3 +1,136 @@
-//! Test helpers that the crates of other Stores share.
+//! The shared suites of the Stream rules. Each Store crate runs them on its own backends, so that
+//! every Store keeps the same rules. This crate runs them on the memory Store.
+
+mod support;
 
 pub mod fields;
+pub mod store;
+
+pub use support::Needs;
+
+use crate::{
+    memory::MemoryStore,
+    store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store},
+};
+use futures_util::{FutureExt, future::join_all};
+use std::{future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc};
+
+/// The state under a Store's calls, which some cases read or change.
+#[async_trait::async_trait]
+pub trait RawKeys: Send + Sync {
+    /// Sets one meta field of a Stream. Field names and values are bytes, as Redis keeps them.
+    async fn set_meta(&self, namespace: &str, stream: &str, field: &[u8], value: &[u8]);
+    /// One meta field of a Stream.
+    async fn meta(&self, namespace: &str, stream: &str, field: &[u8]) -> Option<Vec<u8>>;
+    /// The ms left on the log and on the meta, as `PTTL` gives them.
+    async fn ttls(&self, namespace: &str, stream: &str) -> (i64, i64);
+    /// Adds a log entry with an explicit id, as `XADD <id>`.
+    async fn add_raw_entry(&self, namespace: &str, stream: &str, id: &str);
+    /// The number of log entries.
+    async fn log_len(&self, namespace: &str, stream: &str) -> usize;
+}
+
+/// A backend that the suites run on.
+#[async_trait::async_trait]
+pub trait TestBackend: Send + Sync {
+    /// What the backend gives. A case that needs more doesn't run on it.
+    fn gives(&self) -> Needs;
+    /// A Store for one case, and the state under it.
+    async fn connect(&self) -> (Arc<dyn Store>, Arc<dyn RawKeys>);
+}
+
+/// The memory Store, with a new Store for each case.
+#[derive(Debug, Default)]
+pub struct MemoryBackend;
+
+#[async_trait::async_trait]
+impl TestBackend for MemoryBackend {
+    fn gives(&self) -> Needs {
+        Needs::ALL
+    }
+
+    async fn connect(&self) -> (Arc<dyn Store>, Arc<dyn RawKeys>) {
+        let store = Arc::new(MemoryStore::new(DEFAULT_RETENTION, DEFAULT_SAFETY_TTL));
+        (store.clone(), store)
+    }
+}
+
+#[async_trait::async_trait]
+impl RawKeys for MemoryStore {
+    async fn set_meta(&self, namespace: &str, stream: &str, field: &[u8], value: &[u8]) {
+        self.test_set_meta(namespace, stream, field, value)
+    }
+
+    async fn meta(&self, namespace: &str, stream: &str, field: &[u8]) -> Option<Vec<u8>> {
+        self.test_meta(namespace, stream, field)
+    }
+
+    async fn ttls(&self, namespace: &str, stream: &str) -> (i64, i64) {
+        self.test_ttls(namespace, stream)
+    }
+
+    async fn add_raw_entry(&self, namespace: &str, stream: &str, id: &str) {
+        self.test_add_entry(namespace, stream, id)
+    }
+
+    async fn log_len(&self, namespace: &str, stream: &str) -> usize {
+        self.test_log_len(namespace, stream)
+    }
+}
+
+/// One case of a suite.
+pub struct Case {
+    /// The name of the case, for the report.
+    pub name: &'static str,
+    /// Runs the case on each backend under test that gives what it needs.
+    pub run: fn() -> Pin<Box<dyn Future<Output = ()>>>,
+}
+
+macro_rules! case {
+    ($name:ident) => {
+        $crate::testing::Case {
+            name: stringify!($name),
+            run: || Box::pin($name()),
+        }
+    };
+}
+pub(crate) use case;
+
+tokio::task_local! {
+    static BACKENDS: Vec<Arc<dyn TestBackend>>;
+}
+
+/// The backends of the running suite.
+pub(crate) fn backends() -> Vec<Arc<dyn TestBackend>> {
+    BACKENDS.with(Clone::clone)
+}
+
+/// Runs `cases` on `backends`, all at once, and then fails with the names of the cases that
+/// failed. A case that fails prints its panic as it goes.
+pub async fn run(backends: Vec<Arc<dyn TestBackend>>, cases: &[Case]) {
+    let outcomes = BACKENDS
+        .scope(
+            backends,
+            join_all(cases.iter().map(|case| async move {
+                let passed = AssertUnwindSafe((case.run)()).catch_unwind().await.is_ok();
+                (case.name, passed)
+            })),
+        )
+        .await;
+    let failed: Vec<_> = outcomes
+        .into_iter()
+        .filter(|(_, passed)| !passed)
+        .map(|(name, _)| name)
+        .collect();
+    assert!(failed.is_empty(), "these cases failed: {failed:?}");
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn the_store_suite_passes_on_the_memory_store() {
+        run(vec![Arc::new(MemoryBackend)], store::CASES).await;
+    }
+}
diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs
new file mode 100644
index 000000000..dc9288ec1
--- /dev/null
+++ b/crates/streams/src/testing/store.rs
@@ -0,0 +1,943 @@
+//! The Store suite. Every Store runs it, so that each one keeps the same rules.
+//!
+//! A case says its guarantee and what it needs from a backend in its comment.
+
+use super::{
+    case, fields,
+    support::{
+        Needs, Run, Target, digest, record, records_of, stores_for, superseded, workflow_execution,
+    },
+};
+use crate::{
+    Reason, StreamResult,
+    epoch::{ProducerKind, epoch_key},
+    ids::{
+        activity_producer_id, client_producer_id, named_client_producer_id, workflow_producer_id,
+    },
+    memory::{EntryId, ProducerField},
+    offset::{entry_id_of_offset, offset_of_entry_id},
+    proto::{
+        ActivityEpoch, ActivityProducer, ClientEpoch, ClientProducer, Producer, StreamRef,
+        WorkflowProducer, WorkflowTaskEpoch, producer::Kind,
+    },
+    store::{Batch, ConflictPolicy, Create, CreateOutcome, Created, Owner, Withdrawal},
+};
+use futures_util::future::join_all;
+use prost_types::Timestamp;
+use std::time::Duration;
+
+fn activity(run: &str, attempt: i32, scheduled: i64) -> Producer {
+    Producer {
+        id: activity_producer_id(run, "a", "s"),
+        kind: Some(Kind::Activity(ActivityProducer {
+            run_id: run.to_string(),
+            epoch: Some(ActivityEpoch {
+                attempt_scheduled_time: Some(Timestamp {
+                    seconds: scheduled,
+                    nanos: 0,
+                }),
+                attempt,
+            }),
+        })),
+    }
+}
+
+fn client(attempt: i32) -> Producer {
+    Producer {
+        id: client_producer_id("me", "p1"),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt }),
+        })),
+    }
+}
+
+/// A second client Producer, which every backend accepts as sent.
+fn other_client() -> Producer {
+    Producer {
+        id: named_client_producer_id("other"),
+        ..client(1)
+    }
+}
+
+async fn append(
+    target: &Target,
+    producer: &Producer,
+    sequence: i64,
+    texts: &[&str],
+) -> Result<crate::store::Appended, crate::StreamError> {
+    let records: Vec<_> = texts.iter().map(|text| record(producer, text)).collect();
+    let mut withdrawal = Withdrawal::default();
+    let bytes = superseded(producer, &withdrawal);
+    withdrawal.superseded = &bytes;
+    target
+        .store
+        .append(
+            &target.namespace,
+            &target.stream_ref(),
+            &Batch {
+                producer,
+                sequence,
+                records: &records,
+                digest: digest(texts),
+                withdrawal,
+            },
+        )
+        .await
+}
+
+/// The create of a Stream that uses an existing one.
+fn use_existing() -> Create<'static> {
+    Create {
+        conflict: ConflictPolicy::UseExisting,
+        ..Default::default()
+    }
+}
+
+/// A create writes the key once. A create with another Retention finds the Stream and keeps the
+/// stored Retention. A create with no request id keeps none.
+async fn a_create_writes_the_key_once_and_keeps_its_retention() {
+    for target in stores_for(Needs::STORE_ANSWERS).await {
+        assert_eq!(
+            append(&target, &client(1), 1, &["x"])
+                .await
+                .unwrap_err()
+                .reason,
+            Reason::NotFound
+        );
+        let day = Create {
+            retention: Some(Duration::from_secs(86_400)),
+            ..use_existing()
+        };
+        let created = target.create_with(&day).await.unwrap();
+        assert_eq!(created.outcome, CreateOutcome::Created);
+        let again = target.create_with(&day).await.unwrap();
+        assert_eq!(again.outcome, CreateOutcome::Exists);
+        let other = target.create_with(&use_existing()).await.unwrap();
+        assert_eq!(other.outcome, CreateOutcome::Exists);
+        assert_eq!(other.retention, Duration::from_secs(86_400));
+        if target.gives.gives(Needs::RAW_KEYS) {
+            assert_eq!(target.meta("rid").await, None);
+        }
+    }
+}
+
+/// A create of an open Stream fails as `already_exists`, unless it uses the Stream or it is a
+/// retry of the create that made the Stream. Such a retry has the same request id, and it succeeds
+/// as a create with either policy. The Store keeps that request id as `rid`.
+async fn a_create_of_an_open_stream_fails_unless_it_uses_it_or_retries() {
+    for target in stores_for(Needs::CALLS).await {
+        let first = Create {
+            request_id: "create-1",
+            ..Default::default()
+        };
+        let outcome = |created: StreamResult<Created>| created.unwrap().outcome;
+        let reason = |created: StreamResult<Created>| created.unwrap_err().reason;
+        assert_eq!(
+            outcome(target.create_with(&first).await),
+            CreateOutcome::Created
+        );
+        let other = Create {
+            request_id: "create-2",
+            ..Default::default()
+        };
+        for create in [&other, &Create::default()] {
+            assert_eq!(
+                reason(target.create_with(create).await),
+                Reason::AlreadyExists
+            );
+        }
+        let used = Create {
+            request_id: "create-2",
+            ..use_existing()
+        };
+        assert_eq!(
+            outcome(target.create_with(&used).await),
+            CreateOutcome::Exists
+        );
+        let retry = Create {
+            request_id: "create-1",
+            ..use_existing()
+        };
+        for create in [&first, &retry] {
+            assert_eq!(
+                outcome(target.create_with(create).await),
+                CreateOutcome::Created
+            );
+        }
+        if target.gives.gives(Needs::RAW_KEYS) {
+            assert_eq!(target.meta("rid").await.as_deref(), Some(&b"create-1"[..]));
+        }
+    }
+}
+
+/// A create of an expired, another owner's or a closed Stream fails as `expired`, `not_owner` or
+/// `closed`, in this order. These come before the conflict policy and before a retry of the create
+/// that made the Stream.
+async fn closed_expired_and_not_owner_come_before_already_exists() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        let first = Create {
+            request_id: "create-1",
+            ..Default::default()
+        };
+        target.create_with(&first).await.unwrap();
+        let reason = |created: StreamResult<Created>| created.unwrap_err().reason;
+        let owned = StreamRef {
+            owner: workflow_execution("agent", "first-run"),
+            ..target.stream_ref()
+        };
+        let workflow = Create {
+            owner: Owner::Workflow,
+            ..first.clone()
+        };
+        let not_owner = || target.store.create(&target.namespace, &owned, &workflow);
+        assert_eq!(reason(not_owner().await), Reason::NotOwner);
+        target.close(&client(1)).await.unwrap();
+        assert_eq!(reason(not_owner().await), Reason::NotOwner);
+        for create in [&first, &Create::default()] {
+            assert_eq!(reason(target.create_with(create).await), Reason::Closed);
+        }
+        target.set_meta("exp", "1").await;
+        assert_eq!(reason(not_owner().await), Reason::Expired);
+        assert_eq!(reason(target.create_with(&first).await), Reason::Expired);
+    }
+}
+
+/// A resend with the last sequence gets the offsets of its first result, even after another
+/// Producer wrote, and a lower sequence is refused.
+async fn a_resend_gets_its_first_positions_and_a_lower_sequence_is_refused() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        let first = append(&target, &p, 1, &["p1", "p2"]).await.unwrap();
+        // Another Producer writes between the batch and its resend.
+        append(&target, &other_client(), 1, &["q1"]).await.unwrap();
+        let resend = append(&target, &p, 1, &["p1", "p2"]).await.unwrap();
+        assert_eq!(resend.offsets, first.offsets);
+        if target.gives.gives(Needs::STORE_ANSWERS) {
+            assert!(!first.resend && resend.resend);
+        }
+        append(&target, &p, 2, &["p3"]).await.unwrap();
+        assert_eq!(
+            append(&target, &p, 1, &["p1"]).await.unwrap_err().reason,
+            Reason::SequenceTooLow
+        );
+        let texts = target.texts().await;
+        assert_eq!(texts, ["p1", "p2", "q1", "p3"]);
+    }
+}
+
+/// A resend with the last sequence and other content fails as `seq_divergent` and writes nothing.
+async fn a_resend_with_other_content_fails_as_divergent() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        append(&target, &p, 1, &["p1"]).await.unwrap();
+        let divergent = append(&target, &p, 1, &["other"]).await.unwrap_err();
+        assert_eq!(divergent.reason, Reason::SeqDivergent);
+        assert_eq!(target.texts().await, ["p1"]);
+    }
+}
+
+/// A lower epoch is fenced for each Producer kind.
+async fn a_lower_epoch_is_fenced_for_each_producer_kind() {
+    for target in stores_for(Needs::UNCHECKED).await {
+        target.create(Owner::Workflow).await;
+        let run = Run::new("r1");
+        let la = run.local_activity("k", 8, 1, 2);
+        target.batch(&la, 1, &["la2"]).await.unwrap();
+        let stale = target
+            .batch(&run.local_activity("k", 8, 1, 1), 1, &["la1"])
+            .await;
+        assert_eq!(stale.unwrap_err().reason, Reason::Fenced);
+        append(&target, &activity("r1", 2, 20), 1, &["a2"])
+            .await
+            .unwrap();
+        let stale = append(&target, &activity("r1", 3, 15), 1, &["a3"]).await;
+        assert_eq!(stale.unwrap_err().reason, Reason::Fenced);
+        append(&target, &client(2), 1, &["c2"]).await.unwrap();
+        let stale = append(&target, &client(1), 1, &["c1"]).await;
+        assert_eq!(stale.unwrap_err().reason, Reason::Fenced);
+        assert_eq!(target.texts().await, ["la2", "a2", "c2"]);
+    }
+}
+
+/// A close refuses later appends but not a resend of an earlier batch, and a create of the Stream
+/// fails as closed. With `STORE_ANSWERS`, a second close returns `already`.
+async fn a_close_stops_appends_but_not_a_resend() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        let first = append(&target, &p, 1, &["a"]).await.unwrap();
+        let closed = target.close(&p).await.unwrap();
+        assert_eq!(
+            append(&target, &p, 2, &["b"]).await.unwrap_err().reason,
+            Reason::Closed
+        );
+        assert_eq!(
+            append(&target, &p, 1, &["a"]).await.unwrap().offsets,
+            first.offsets
+        );
+        let again = target.close(&p).await.unwrap();
+        if target.gives.gives(Needs::STORE_ANSWERS) {
+            assert!(!closed.already && again.already);
+        }
+        let created = target.create_with(&use_existing()).await;
+        assert_eq!(created.unwrap_err().reason, Reason::Closed);
+    }
+}
+
+/// A Stream with a newer format version refuses every write. A write checks the version before it
+/// reads the Producer's `p|` field, since a newer format can change that field.
+async fn a_newer_format_refuses_every_write() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        append(&target, &p, 1, &["a"]).await.unwrap();
+        target.set_meta(fields::VERSION, "2").await;
+        let field = fields::producer(ProducerKind::Client as i32, &p.id);
+        target.set_meta(&field, "a field of a newer format").await;
+        let error = append(&target, &p, 2, &["b"]).await.unwrap_err();
+        assert_eq!(error.reason, Reason::FormatTooNew);
+        assert_eq!(
+            target.close(&p).await.unwrap_err().reason,
+            Reason::FormatTooNew
+        );
+        let create = target.create_with(&use_existing()).await;
+        assert_eq!(create.unwrap_err().reason, Reason::FormatTooNew);
+    }
+}
+
+/// An expired Stream refuses writes and its create.
+async fn an_expired_stream_refuses_writes_and_its_create() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        target.set_meta("exp", "1").await;
+        let error = append(&target, &client(1), 1, &["a"]).await.unwrap_err();
+        assert_eq!(error.reason, Reason::Expired);
+        let created = target.create_with(&use_existing()).await;
+        assert_eq!(created.unwrap_err().reason, Reason::Expired);
+    }
+}
+
+/// An Activity retry flags only its first batch, which withdraws the earlier attempt.
+async fn an_activity_retry_flags_its_first_batch_only() {
+    for target in stores_for(Needs::UNCHECKED.and(Needs::STORE_ANSWERS)).await {
+        target.create(Owner::Workflow).await;
+        let (a1, a2) = (activity("r1", 1, 10), activity("r1", 2, 20));
+        let other = Producer {
+            id: activity_producer_id("r1", "b", "s"),
+            ..activity("r1", 1, 10)
+        };
+        append(&target, &a1, 1, &["x1"]).await.unwrap();
+        append(&target, &other, 1, &["y1"]).await.unwrap();
+        assert!(append(&target, &a2, 1, &["x2"]).await.unwrap().flagged);
+        assert!(!append(&target, &a2, 2, &["x3"]).await.unwrap().flagged);
+        assert_eq!(target.visible().await, ["y1", "x2", "x3"]);
+    }
+}
+
+/// An empty batch of an Activity retry withdraws the earlier attempt's Records.
+async fn an_empty_batch_of_a_retry_withdraws_the_earlier_attempt() {
+    for target in stores_for(Needs::UNCHECKED).await {
+        target.create(Owner::Workflow).await;
+        append(&target, &activity("r1", 1, 10), 1, &["x1"])
+            .await
+            .unwrap();
+        let empty = append(&target, &activity("r1", 2, 20), 1, &[])
+            .await
+            .unwrap();
+        assert!(empty.flagged);
+        assert!(empty.offsets.is_some());
+        assert!(target.visible().await.is_empty());
+        // The withdrawal is done, so another empty batch writes nothing.
+        let again = append(&target, &activity("r1", 3, 30), 1, &[])
+            .await
+            .unwrap();
+        assert!(!again.flagged);
+        assert_eq!(again.offsets, None);
+        // An empty batch still raises the fence.
+        let stale = append(&target, &activity("r1", 2, 20), 2, &["late"]).await;
+        assert_eq!(stale.unwrap_err().reason, Reason::Fenced);
+    }
+}
+
+/// An open Workflow Stream's meta keeps the safety TTL, and its log keeps the Retention, after
+/// which a read hides every Record. Its close gives the meta the Retention too.
+async fn the_ttl_follows_the_owner_and_the_close() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Workflow).await;
+        let run = Run::new("r1");
+        target
+            .batch(&run.local_activity("k", 3, 1, 1), 1, &["a"])
+            .await
+            .unwrap();
+        let week = Duration::from_secs(7 * 86_400).as_millis() as i64;
+        let (ttl, meta) = target.ttls().await;
+        assert!(0 < ttl && ttl <= week, "{ttl}");
+        assert!(
+            meta > Duration::from_secs(80 * 86_400).as_millis() as i64,
+            "{meta}"
+        );
+        target.close(&workflow("r1")).await.unwrap();
+        let (ttl, meta) = target.ttls().await;
+        assert!(0 < ttl && ttl <= week, "{ttl}");
+        assert!(
+            meta > Duration::from_secs(36 * 86_400).as_millis() as i64,
+            "{meta}"
+        );
+    }
+}
+
+/// A read hides Records past the Retention, and the next append trims them and keeps the
+/// newest trimmed offset.
+async fn an_append_trims_by_retention_and_keeps_the_watermark() {
+    for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
+        target
+            .create_with(&Create {
+                retention: Some(Duration::from_millis(600)),
+                owner: Owner::Workflow,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        let old = append(&target, &client(1), 1, &["old"]).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(400)).await;
+        // This append keeps the log past the old Record's Retention.
+        append(&target, &client(1), 2, &["mid"]).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(350)).await;
+        // The read hides the old Record before any append trims it.
+        let (state, entries) = target
+            .store
+            .read(&target.namespace, &target.stream, 0, 10)
+            .await
+            .unwrap();
+        assert!(state.retained_from() > old.offsets.unwrap().1);
+        assert_eq!(entries.len(), 2);
+        append(&target, &client(1), 3, &["new"]).await.unwrap();
+        let (state, entries) = target
+            .store
+            .read(&target.namespace, &target.stream, 0, 10)
+            .await
+            .unwrap();
+        assert_eq!(state.trimmed, old.offsets.unwrap().1);
+        assert_eq!(records_of(&entries), ["mid", "new"]);
+    }
+}
+
+/// The log expires one Retention after the last write. The next write raises the trimmed offset
+/// to the meta's newest one, and its offsets go above it, also with the clock behind it.
+async fn offsets_go_on_after_the_log_expires() {
+    for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
+        target
+            .create_with(&Create {
+                retention: Some(Duration::from_millis(200)),
+                owner: Owner::Workflow,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        append(&target, &client(1), 1, &["old"]).await.unwrap();
+        // A newest id one minute ahead, as after the clock steps back.
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis();
+        let ahead = format!("{}-3", now + 60_000);
+        target.set_meta("newest", &ahead).await;
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(target.ttls().await.0, -2);
+        let new = append(&target, &client(1), 2, &["new"]).await.unwrap();
+        let ahead = offset_of_entry_id(&ahead).unwrap();
+        assert!(new.offsets.unwrap().0 > ahead);
+        let (state, _) = target
+            .store
+            .read(&target.namespace, &target.stream, 0, 0)
+            .await
+            .unwrap();
+        assert_eq!(state.trimmed, ahead);
+        assert_eq!(target.texts().await, ["new"]);
+    }
+}
+
+/// A wait for the next entry wakes on an append and ends at its deadline.
+async fn a_blocked_read_wakes_on_an_append_and_ends_at_its_deadline() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        let first = append(&target, &client(1), 1, &["a"]).await.unwrap();
+        let newest = first.offsets.unwrap().1;
+        let deadline = tokio::time::Instant::now() + Duration::from_millis(200);
+        let woke = target
+            .store
+            .wait_after(&target.namespace, &target.stream, newest, deadline)
+            .await
+            .unwrap();
+        assert!(!woke);
+        let store = target.store.clone();
+        let (namespace, stream) = (target.namespace.clone(), target.stream.clone());
+        let waiter = tokio::spawn(async move {
+            let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+            store
+                .wait_after(&namespace, &stream, newest, deadline)
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        append(&target, &client(1), 2, &["b"]).await.unwrap();
+        assert!(waiter.await.unwrap().unwrap());
+    }
+}
+
+/// A close writes no log entry. It ends a wait for the next entry, and a wait on a closed Stream
+/// ends at once, so a waiting read sees the close.
+async fn a_close_writes_no_entry_and_ends_a_wait() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        let first = append(&target, &client(1), 1, &["a"]).await.unwrap();
+        let newest = first.offsets.unwrap().1;
+        let store = target.store.clone();
+        let (namespace, stream) = (target.namespace.clone(), target.stream.clone());
+        let waiter = tokio::spawn(async move {
+            let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+            store
+                .wait_after(&namespace, &stream, newest, deadline)
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        target.close(&client(1)).await.unwrap();
+        assert!(waiter.await.unwrap().unwrap());
+        let started = tokio::time::Instant::now();
+        let deadline = started + Duration::from_secs(5);
+        let woke = target
+            .store
+            .wait_after(&target.namespace, &target.stream, newest, deadline)
+            .await
+            .unwrap();
+        assert!(woke && started.elapsed() < Duration::from_secs(1));
+        let (state, entries) = target
+            .store
+            .read(&target.namespace, &target.stream, 0, 10)
+            .await
+            .unwrap();
+        assert!(state.closed);
+        assert_eq!((state.newest, entries.len()), (newest, 1));
+    }
+}
+
+/// The Records of one batch get consecutive offsets that end at the batch's last offset, so a
+/// caller knows the offset of each one.
+async fn a_batch_gets_consecutive_offsets_that_end_at_its_last_offset() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        append(&target, &client(1), 1, &["before"]).await.unwrap();
+        // Enough Records that `XADD *` on Redis 7.0 would cross a millisecond.
+        let texts: Vec<String> = (0..1000).map(|i| i.to_string()).collect();
+        let texts: Vec<&str> = texts.iter().map(String::as_str).collect();
+        let appended = append(&target, &client(1), 2, &texts).await.unwrap();
+        let offsets: Vec<i64> = target.entries().await[1..]
+            .iter()
+            .map(|entry| entry.offset)
+            .collect();
+        assert_eq!(offsets.len(), texts.len());
+        assert_eq!(
+            appended.offsets,
+            Some((offsets[0], offsets[offsets.len() - 1]))
+        );
+        assert!(
+            offsets.windows(2).all(|pair| pair[1] == pair[0] + 1),
+            "{offsets:?}"
+        );
+    }
+}
+
+/// A negative sequence is refused.
+async fn a_negative_sequence_is_refused() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let error = append(&target, &client(1), -1, &["x"]).await.unwrap_err();
+        assert_eq!(error.reason, Reason::Invalid);
+        assert!(target.texts().await.is_empty());
+    }
+}
+
+/// A batch with no sequence takes the next one of its Producer's epoch, and a newer epoch starts
+/// at one. Such a batch has no dedupe, so its retry writes it again. A batch that sends the taken
+/// number is a resend.
+async fn a_batch_with_no_sequence_takes_the_next_one_of_its_epoch() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        append(&target, &p, 0, &["a"]).await.unwrap();
+        let retry = append(&target, &p, 0, &["a"]).await.unwrap();
+        let resend = append(&target, &p, 2, &["a"]).await.unwrap();
+        assert_eq!(resend.offsets, retry.offsets);
+        let lower = append(&target, &p, 1, &["a"]).await.unwrap_err();
+        assert_eq!(lower.reason, Reason::SequenceTooLow);
+        append(&target, &p, 7, &["b"]).await.unwrap();
+        let next = append(&target, &p, 0, &["c"]).await.unwrap();
+        let resend = append(&target, &p, 8, &["c"]).await.unwrap();
+        assert_eq!(resend.offsets, next.offsets);
+        let newer = client(2);
+        let first = append(&target, &newer, 0, &["d"]).await.unwrap();
+        let resend = append(&target, &newer, 1, &["d"]).await.unwrap();
+        assert_eq!(resend.offsets, first.offsets);
+        assert_eq!(target.texts().await, ["a", "a", "b", "c", "d"]);
+    }
+}
+
+/// Batches with no sequence that one Producer epoch sends at once each take their own number, so
+/// the Store writes all of them.
+async fn batches_with_no_sequence_at_once_each_take_their_own_number() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        let texts = ["t1", "t2", "t3", "t4", "t5", "t6", "t7", "t8"];
+        let appends = texts
+            .iter()
+            .map(|text| append(&target, &p, 0, std::slice::from_ref(text)));
+        for appended in join_all(appends).await {
+            appended.unwrap();
+        }
+        let mut written = target.texts().await;
+        written.sort();
+        assert_eq!(written, texts);
+        // The batches took 1 to 8, so 8 holds one of them.
+        let taken = append(&target, &p, 8, &["other"]).await.unwrap_err();
+        assert_eq!(taken.reason, Reason::SeqDivergent);
+    }
+}
+
+/// A large sequence keeps its dedupe entry, so a resend gets its first answer and a lower or
+/// divergent batch is refused. A sequence past 2^53 - 1 is refused.
+async fn a_large_sequence_dedupes_and_one_past_2_53_is_refused() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        let large = 100_000_000_000_000;
+        let first = append(&target, &p, large, &["x"]).await.unwrap();
+        let resend = append(&target, &p, large, &["x"]).await.unwrap();
+        assert_eq!(resend.offsets, first.offsets);
+        let divergent = append(&target, &p, large, &["other"]).await;
+        assert_eq!(divergent.unwrap_err().reason, Reason::SeqDivergent);
+        let lower = append(&target, &p, large - 1, &["lower"]).await;
+        assert_eq!(lower.unwrap_err().reason, Reason::SequenceTooLow);
+        assert_eq!(target.texts().await, ["x"]);
+        let max = (1 << 53) - 1;
+        append(&target, &p, max, &["max"]).await.unwrap();
+        let past = append(&target, &p, max + 1, &["past"]).await;
+        assert_eq!(past.unwrap_err().reason, Reason::Invalid);
+    }
+}
+
+/// A create that names another owner than the Stream has fails and changes nothing. So a Workflow
+/// doesn't take a Stream with no owner, and another run chain of a Workflow id is another owner.
+async fn a_create_with_another_owner_is_refused() {
+    for target in stores_for(Needs::RAW_KEYS.and(Needs::STORE_ANSWERS)).await {
+        let target = &target;
+        let create = |stream_ref: StreamRef, owner: Owner| async move {
+            let create = Create {
+                owner,
+                ..use_existing()
+            };
+            target
+                .store
+                .create(&target.namespace, &stream_ref, &create)
+                .await
+        };
+        let owned = |workflow_id: &str, first_run_id: &str| StreamRef {
+            owner: workflow_execution(workflow_id, first_run_id),
+            ..target.stream_ref()
+        };
+        target.create(Owner::Client).await;
+        let refused = create(owned("agent", "first-run"), Owner::Workflow).await;
+        assert_eq!(refused.unwrap_err().reason, Reason::NotOwner);
+        assert_eq!(target.meta("own").await.as_deref(), Some(&b"client"[..]));
+        // The meta keeps the TTL of a client's Stream, below the safety TTL of 90 days.
+        let (_, ttl) = target.ttls().await;
+        assert!(
+            ttl < Duration::from_secs(90 * 86_400).as_millis() as i64,
+            "{ttl}"
+        );
+
+        let owned = |workflow_id: &str, first_run_id: &str| StreamRef {
+            id: format!("{}-owned", target.stream),
+            ..owned(workflow_id, first_run_id)
+        };
+        let created = create(owned("agent", "first-run"), Owner::Workflow).await;
+        assert_eq!(created.unwrap().outcome, CreateOutcome::Created);
+        for (stream, owner) in [
+            (owned("agent", "other-run"), Owner::Workflow),
+            (owned("another-agent", "first-run"), Owner::Workflow),
+            (owned("agent", "first-run"), Owner::Client),
+        ] {
+            let refused = create(stream, owner).await;
+            assert_eq!(refused.unwrap_err().reason, Reason::NotOwner);
+        }
+        let again = create(owned("agent", "first-run"), Owner::Workflow).await;
+        assert_eq!(again.unwrap().outcome, CreateOutcome::Exists);
+    }
+}
+
+/// An append removes the `p|` field of a Producer whose last append is older than the Retention,
+/// since its Records are gone too. A lower epoch of that Producer then writes.
+async fn an_append_forgets_the_producers_that_the_retention_outlived() {
+    for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
+        target
+            .create_with(&Create {
+                retention: Some(Duration::from_millis(300)),
+                owner: Owner::Workflow,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        let (old, new) = (client(2), other_client());
+        append(&target, &old, 1, &["old"]).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(400)).await;
+        append(&target, &new, 1, &["new"]).await.unwrap();
+        let kind = ProducerKind::Client as i32;
+        assert_eq!(target.meta(fields::producer(kind, &old.id)).await, None);
+        assert!(target.meta(fields::producer(kind, &new.id)).await.is_some());
+        append(&target, &client(1), 1, &["late"]).await.unwrap();
+        assert_eq!(target.texts().await, ["new", "late"]);
+    }
+}
+
+/// A Producer with a lower epoch can't close the Stream.
+async fn a_stale_producer_cant_close_the_stream() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        append(&target, &client(2), 1, &["a"]).await.unwrap();
+        assert_eq!(
+            target.close(&client(1)).await.unwrap_err().reason,
+            Reason::Fenced
+        );
+        target.close(&client(2)).await.unwrap();
+    }
+}
+
+/// The close Activity's Producer of run `run`.
+fn workflow(run: &str) -> Producer {
+    Producer {
+        id: workflow_producer_id(run),
+        kind: Some(Kind::Workflow(WorkflowProducer {
+            run_id: run.to_string(),
+            epoch: Some(WorkflowTaskEpoch::default()),
+        })),
+    }
+}
+
+/// A Stream with no owner takes a close from a Workflow or a client. A Local Activity or an
+/// Activity can't close it.
+async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() {
+    for target in stores_for(Needs::UNCHECKED).await {
+        target.create(Owner::Client).await;
+        let la = Run::new("r1").local_activity("k", 3, 1, 1);
+        for producer in [&la, &activity("r1", 1, 10)] {
+            let refused = target.close(producer).await.unwrap_err();
+            assert_eq!(refused.reason, Reason::NotOwner);
+        }
+        target.close(&workflow("r1")).await.unwrap();
+    }
+}
+
+/// A Store refuses an append or a close of a Producer with no epoch, or with an epoch that has no
+/// case, since the case gives the kind of the Producer.
+async fn a_producer_with_no_epoch_case_is_refused() {
+    for target in stores_for(Needs::CALLS).await {
+        target.create(Owner::Client).await;
+        let producer = Producer {
+            kind: None,
+            ..client(1)
+        };
+        let error = append(&target, &producer, 1, &["a"]).await.unwrap_err();
+        assert_eq!(error.reason, Reason::Invalid);
+        let error = target.close(&producer).await.unwrap_err();
+        assert_eq!(error.reason, Reason::Invalid);
+        assert!(target.texts().await.is_empty());
+    }
+}
+
+/// Only the owner closes a Stream. That's the close Activity of its run chain for a Workflow-owned
+/// Stream, and a Workflow or a client for a Stream with no owner. A Local Activity of the chain
+/// can't close it, and a ref with no chain can't either.
+async fn only_the_owner_closes_a_stream() {
+    for target in stores_for(Needs::UNCHECKED.and(Needs::STORE_ANSWERS)).await {
+        let run = Run::new("r1");
+        let la = run.local_activity("k", 3, 1, 1);
+        target.create(Owner::Client).await;
+        for producer in [&la, &activity("r1", 1, 10)] {
+            let refused = target.close(producer).await.unwrap_err();
+            assert_eq!(refused.reason, Reason::NotOwner);
+        }
+        target.close(&client(1)).await.unwrap();
+
+        let owned = |workflow_id: &str| StreamRef {
+            id: format!("{}-owned", target.stream),
+            owner: workflow_execution(workflow_id, "first-run").filter(|_| !workflow_id.is_empty()),
+            ..Default::default()
+        };
+        let create = Create {
+            owner: Owner::Workflow,
+            ..Default::default()
+        };
+        target
+            .store
+            .create(&target.namespace, &owned("agent"), &create)
+            .await
+            .unwrap();
+        for (stream, producer) in [
+            (owned("agent"), client(1)),
+            (owned("agent"), activity("r1", 1, 10)),
+            (owned("agent"), la.clone()),
+            (owned("another-agent"), workflow("r1")),
+            (owned(""), workflow("r1")),
+        ] {
+            let refused = target.close_at(&stream, &producer).await.unwrap_err();
+            assert_eq!(refused.reason, Reason::NotOwner);
+        }
+        let closed = target.close_at(&owned("agent"), &workflow("r1")).await;
+        assert!(!closed.unwrap().already);
+        // The owner check comes first, even on a closed Stream.
+        let refused = target.close_at(&owned("agent"), &la).await.unwrap_err();
+        assert_eq!(refused.reason, Reason::NotOwner);
+    }
+}
+
+/// An append whose entry ids would not fit an offset is refused before it writes.
+async fn an_append_whose_ids_would_not_fit_an_offset_is_refused_before_it_writes() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        // An entry a minute ahead of the clock, with nearly the last 20-bit sequence.
+        let ahead = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis()
+            + 60_000;
+        target.add_raw_entry(&format!("{ahead}-1048570")).await;
+        let error = append(&target, &client(1), 1, &["a"; 10])
+            .await
+            .unwrap_err();
+        assert_eq!(error.reason, Reason::StorageUnavailable);
+        assert_eq!(target.log_len().await, 1);
+    }
+}
+
+/// Every Store keeps the meta fields that the shared vectors name.
+async fn every_store_keeps_the_meta_fields_of_the_shared_vectors() {
+    let vectors: serde_json::Value =
+        serde_json::from_str(include_str!("../../testdata/fields.json")).unwrap();
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Workflow).await;
+        let producer = activity("run:1", 1, 10);
+        append(&target, &producer, 1, &["a"]).await.unwrap();
+        let kind = ProducerKind::Activity as i32;
+        // The vector's Producer is this one, so its field names are the ones the Store keeps.
+        assert_eq!(producer.id, vectors["producer_id"].as_str().unwrap());
+        let field = fields::producer(kind, &producer.id);
+        assert_eq!(field, vectors["producer"].as_str().unwrap());
+        for field in [field, fields::VERSION.to_string()] {
+            assert!(target.meta(&field).await.is_some(), "{field}");
+        }
+    }
+}
+
+/// The entry ids of an append's offsets.
+fn entry_ids(offsets: Option<(i64, i64)>) -> Option<(EntryId, EntryId)> {
+    let id = |offset: i64| {
+        let text = entry_id_of_offset(offset);
+        let (ms, seq) = text.split_once('-').unwrap();
+        (ms.parse().unwrap(), seq.parse().unwrap())
+    };
+    offsets.map(|(first, last)| (id(first), id(last)))
+}
+
+/// A Producer's `p|` field has the same 129 bytes on every Store. These are the flags, the fence,
+/// the held epoch, and the last batch's sequence, entry ids and digest.
+async fn a_producer_field_has_the_shared_layout() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Workflow).await;
+        let kind = ProducerKind::Activity as i32;
+        let field = |producer: &Producer, held: Option<&Producer>, offsets, texts: &[&str]| {
+            let epoch = |producer: &Producer| epoch_key(producer.kind.as_ref());
+            ProducerField {
+                fence: epoch(producer),
+                held: held.map(epoch),
+                sequence: 1,
+                entries: entry_ids(offsets),
+                digest: digest(texts),
+            }
+            .encode()
+        };
+        let a1 = activity("r1", 1, 10);
+        let stored = || async { target.meta(fields::producer(kind, &a1.id)).await.unwrap() };
+        let written = append(&target, &a1, 1, &["x", "y"]).await.unwrap();
+        assert_eq!(
+            stored().await,
+            field(&a1, Some(&a1), written.offsets, &["x", "y"])
+        );
+        // A newer attempt's empty batch withdraws the held epoch's Records, so none is held.
+        let a2 = activity("r1", 2, 20);
+        let flagged = append(&target, &a2, 1, &[]).await.unwrap();
+        assert_eq!(stored().await, field(&a2, None, flagged.offsets, &[]));
+        // Nothing is left to withdraw, so the next empty batch writes no entry.
+        let a3 = activity("r1", 3, 30);
+        let empty = append(&target, &a3, 1, &[]).await.unwrap();
+        assert_eq!(empty.offsets, None);
+        assert_eq!(stored().await, field(&a3, None, None, &[]));
+    }
+}
+
+/// A Store refuses a write of a Producer whose `p|` field has another length or another flag bit,
+/// which a later format may write.
+async fn a_producer_field_of_another_format_is_refused() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create(Owner::Client).await;
+        let p = client(1);
+        append(&target, &p, 1, &["a"]).await.unwrap();
+        let field = fields::producer(ProducerKind::Client as i32, &p.id);
+        let value = target.meta(&field).await.unwrap();
+        let mut flagged = value.clone();
+        flagged[0] |= 2;
+        for other in [flagged, value[..value.len() - 1].to_vec()] {
+            target.set_meta(&field, &other).await;
+            let error = append(&target, &p, 2, &["b"]).await.unwrap_err();
+            assert_eq!(error.reason, Reason::StorageCorrupt);
+            let error = target.close(&p).await.unwrap_err();
+            assert_eq!(error.reason, Reason::StorageCorrupt);
+        }
+        assert_eq!(target.texts().await, ["a"]);
+    }
+}
+
+/// The cases of the suite, in file order.
+pub const CASES: &[super::Case] = &[
+    case!(a_create_writes_the_key_once_and_keeps_its_retention),
+    case!(a_create_of_an_open_stream_fails_unless_it_uses_it_or_retries),
+    case!(closed_expired_and_not_owner_come_before_already_exists),
+    case!(a_resend_gets_its_first_positions_and_a_lower_sequence_is_refused),
+    case!(a_resend_with_other_content_fails_as_divergent),
+    case!(a_lower_epoch_is_fenced_for_each_producer_kind),
+    case!(a_close_stops_appends_but_not_a_resend),
+    case!(a_newer_format_refuses_every_write),
+    case!(an_expired_stream_refuses_writes_and_its_create),
+    case!(an_activity_retry_flags_its_first_batch_only),
+    case!(an_empty_batch_of_a_retry_withdraws_the_earlier_attempt),
+    case!(the_ttl_follows_the_owner_and_the_close),
+    case!(an_append_trims_by_retention_and_keeps_the_watermark),
+    case!(offsets_go_on_after_the_log_expires),
+    case!(a_blocked_read_wakes_on_an_append_and_ends_at_its_deadline),
+    case!(a_close_writes_no_entry_and_ends_a_wait),
+    case!(a_batch_gets_consecutive_offsets_that_end_at_its_last_offset),
+    case!(a_negative_sequence_is_refused),
+    case!(a_batch_with_no_sequence_takes_the_next_one_of_its_epoch),
+    case!(batches_with_no_sequence_at_once_each_take_their_own_number),
+    case!(a_large_sequence_dedupes_and_one_past_2_53_is_refused),
+    case!(a_create_with_another_owner_is_refused),
+    case!(an_append_forgets_the_producers_that_the_retention_outlived),
+    case!(a_stale_producer_cant_close_the_stream),
+    case!(a_workflow_or_a_client_closes_a_stream_with_no_owner),
+    case!(a_producer_with_no_epoch_case_is_refused),
+    case!(only_the_owner_closes_a_stream),
+    case!(an_append_whose_ids_would_not_fit_an_offset_is_refused_before_it_writes),
+    case!(every_store_keeps_the_meta_fields_of_the_shared_vectors),
+    case!(a_producer_field_has_the_shared_layout),
+    case!(a_producer_field_of_another_format_is_refused),
+];
diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs
new file mode 100644
index 000000000..0426f0bcf
--- /dev/null
+++ b/crates/streams/src/testing/support.rs
@@ -0,0 +1,365 @@
+//! The helpers that the suites share. They give a new Stream per case on each backend under test,
+//! and the Producers of a Workflow run.
+
+#![allow(dead_code)]
+
+use super::{RawKeys, backends};
+use crate::{
+    StreamResult,
+    digest::batch_digest,
+    epoch::withdraws,
+    ids::local_activity_producer_id,
+    proto::{
+        AppendRecord, LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal,
+        StreamData, StreamRecord, StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch,
+        producer::Kind, stream_record::Item,
+    },
+    store::{Appended, Batch, Closed, Create, Created, Entry, Owner, Store, Withdrawal},
+};
+use prost::Message;
+use prost_types::Timestamp;
+use std::sync::{
+    Arc,
+    atomic::{AtomicU64, Ordering},
+};
+use temporalio_common::protos::temporal::api::{common::v1::Execution, enums::v1::ExecutionType};
+
+static NEXT: AtomicU64 = AtomicU64::new(0);
+
+/// What a case needs beyond the Store's calls on its own Stream. A case runs on every backend that
+/// has all that it needs. Each case's comment says its guarantee and its needs.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct Needs(u8);
+
+impl Needs {
+    /// Only the Store's calls on the Stream and reads of its Records.
+    pub const CALLS: Self = Self(0);
+    /// The Store's own results beyond the API. These are a create's stored Retention, `resend`,
+    /// `flagged`, and a close's offset and `already`.
+    pub const STORE_ANSWERS: Self = Self(1);
+    /// The keys under the calls. These are the meta fields, TTLs and raw log entries, and the
+    /// Store's log calls `read`, `read_back` and `wait_after`.
+    pub const RAW_KEYS: Self = Self(1 << 1);
+    /// Producers, owners and runs as sent. The server checks them against its executions, so a
+    /// case with made-up runs, or an Activity with no task token, can't run there.
+    pub const UNCHECKED: Self = Self(1 << 2);
+    /// A Retention of a few hundred ms.
+    pub const SHORT_TIMES: Self = Self(1 << 3);
+    /// The runtime built from the crate's own config, such as a backend name.
+    pub const OWN_CONFIG: Self = Self(1 << 4);
+    /// The Store keys on the id only, so the owner of a ref doesn't route a call.
+    pub const ID_KEYS: Self = Self(1 << 5);
+    /// Every need. Redis and memory have all of them.
+    pub const ALL: Self = Self(u8::MAX);
+
+    /// Both needs.
+    pub const fn and(self, other: Self) -> Self {
+        Self(self.0 | other.0)
+    }
+
+    /// Whether this set gives every need of `needs`.
+    pub const fn gives(self, needs: Self) -> bool {
+        self.0 & needs.0 == needs.0
+    }
+}
+
+/// One Stream on one server, fresh for each case.
+pub(crate) struct Target {
+    pub(crate) store: Arc<dyn Store>,
+    pub(crate) raw: Arc<dyn RawKeys>,
+    pub(crate) namespace: String,
+    pub(crate) stream: String,
+    /// What the backend gives. A case checks it before a step that needs more than it asked for.
+    pub(crate) gives: Needs,
+}
+
+fn unique() -> String {
+    let n = NEXT.fetch_add(1, Ordering::Relaxed);
+    format!(
+        "{}-{n}",
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    )
+}
+
+/// A new Stream on each backend under test that gives `needs`.
+pub(crate) async fn stores_for(needs: Needs) -> Vec<Target> {
+    let mut targets = vec![];
+    for backend in backends() {
+        if !backend.gives().gives(needs) {
+            continue;
+        }
+        let (store, raw) = backend.connect().await;
+        let unique = unique();
+        targets.push(Target {
+            store,
+            raw,
+            namespace: format!("ns-{unique}"),
+            stream: format!("wf:{unique}"),
+            gives: backend.gives(),
+        });
+    }
+    targets
+}
+
+/// The Producers of one Workflow run.
+pub(crate) struct Run {
+    pub(crate) id: String,
+}
+
+fn at(seconds: i64) -> Option<Timestamp> {
+    Some(Timestamp { seconds, nanos: 0 })
+}
+
+impl Run {
+    pub(crate) fn new(id: &str) -> Self {
+        Self { id: id.to_string() }
+    }
+
+    pub(crate) fn local_activity(
+        &self,
+        call: &str,
+        started: i64,
+        task_attempt: i32,
+        attempt: i32,
+    ) -> Producer {
+        let task = WorkflowTaskEpoch {
+            started_event_id: started,
+            attempt: task_attempt,
+            started_time: at(100),
+        };
+        Producer {
+            id: local_activity_producer_id(&self.id, call),
+            kind: Some(Kind::LocalActivity(LocalActivityProducer {
+                run_id: self.id.clone(),
+                epoch: Some(LocalActivityEpoch {
+                    task: Some(task),
+                    attempt,
+                }),
+            })),
+        }
+    }
+}
+
+/// The owner of a ref that a Workflow's run chain makes, with the chain's first run id.
+pub(crate) fn workflow_execution(workflow_id: &str, first_run_id: &str) -> Option<Execution> {
+    Some(Execution {
+        r#type: ExecutionType::Workflow as i32,
+        business_id: workflow_id.to_string(),
+        run_id: first_run_id.to_string(),
+    })
+}
+
+pub(crate) fn record(producer: &Producer, text: &str) -> Vec<u8> {
+    StreamRecord {
+        item: Some(Item::Data(StreamData {
+            producer: Some(producer.clone()),
+            body: Some(
+                temporalio_common::protos::temporal::api::common::v1::Payload {
+                    data: text.as_bytes().to_vec(),
+                    ..Default::default()
+                },
+            ),
+        })),
+        ..Default::default()
+    }
+    .encode_to_vec()
+}
+
+/// The digest of the batch that [record] makes of each text.
+pub(crate) fn digest(texts: &[&str]) -> [u8; crate::digest::DIGEST_LEN] {
+    let records: Vec<_> = texts
+        .iter()
+        .map(|text| AppendRecord {
+            body: Some(
+                temporalio_common::protos::temporal::api::common::v1::Payload {
+                    data: text.as_bytes().to_vec(),
+                    ..Default::default()
+                },
+            ),
+        })
+        .collect();
+    batch_digest(&records)
+}
+
+pub(crate) fn superseded(producer: &Producer, withdrawal: &Withdrawal<'_>) -> Vec<u8> {
+    let run = (!withdrawal.withdraw_run.is_empty()).then(|| RunWithdrawal {
+        withdraw_run: withdrawal.withdraw_run.to_string(),
+        ..Default::default()
+    });
+    StreamSuperseded {
+        writer: Some(StreamWriter {
+            producer: Some(producer.clone()),
+            withdrawal: run,
+        }),
+    }
+    .encode_to_vec()
+}
+
+/// The data texts of the entries, in log order.
+pub(crate) fn records_of(entries: &[Entry]) -> Vec<String> {
+    entries
+        .iter()
+        .filter_map(|entry| entry.record.as_ref())
+        .map(|bytes| StreamRecord::decode(bytes.as_slice()).unwrap())
+        .filter_map(|record| match record.item {
+            Some(Item::Data(data)) => Some(String::from_utf8(data.body.unwrap().data).unwrap()),
+            _ => None,
+        })
+        .collect()
+}
+
+impl Target {
+    pub(crate) fn stream_ref(&self) -> StreamRef {
+        StreamRef {
+            id: self.stream.clone(),
+            ..Default::default()
+        }
+    }
+
+    pub(crate) async fn create(&self, owner: Owner) {
+        self.create_with(&Create {
+            owner,
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    }
+
+    /// Creates the Stream, and gives the Store's answer.
+    pub(crate) async fn create_with(&self, create: &Create<'_>) -> StreamResult<Created> {
+        self.store
+            .create(&self.namespace, &self.stream_ref(), create)
+            .await
+    }
+
+    /// Sets one meta field under the Store's calls.
+    pub(crate) async fn set_meta(&self, field: impl AsRef<[u8]>, value: impl AsRef<[u8]>) {
+        self.raw
+            .set_meta(
+                &self.namespace,
+                &self.stream,
+                field.as_ref(),
+                value.as_ref(),
+            )
+            .await
+    }
+
+    /// One meta field.
+    pub(crate) async fn meta(&self, field: impl AsRef<[u8]>) -> Option<Vec<u8>> {
+        self.raw
+            .meta(&self.namespace, &self.stream, field.as_ref())
+            .await
+    }
+
+    /// The ms left on the log and on the meta, as `PTTL` gives them.
+    pub(crate) async fn ttls(&self) -> (i64, i64) {
+        self.raw.ttls(&self.namespace, &self.stream).await
+    }
+
+    /// Adds a log entry with an explicit id, as `XADD <id>`.
+    pub(crate) async fn add_raw_entry(&self, id: &str) {
+        self.raw
+            .add_raw_entry(&self.namespace, &self.stream, id)
+            .await
+    }
+
+    /// The number of log entries.
+    pub(crate) async fn log_len(&self) -> usize {
+        self.raw.log_len(&self.namespace, &self.stream).await
+    }
+
+    /// Every entry of the log.
+    pub(crate) async fn entries(&self) -> Vec<Entry> {
+        let (_, entries) = self
+            .store
+            .read(&self.namespace, &self.stream, 0, 10_000)
+            .await
+            .unwrap();
+        entries
+    }
+
+    /// The data texts in the log, withdrawn or not.
+    pub(crate) async fn texts(&self) -> Vec<String> {
+        records_of(&self.entries().await)
+    }
+
+    /// The data texts a Reader keeps after it applies each `Superseded` item.
+    pub(crate) async fn visible(&self) -> Vec<String> {
+        let mut kept: Vec<StreamRecord> = vec![];
+        for entry in self.entries().await {
+            if let Some(bytes) = &entry.superseded {
+                let s = StreamSuperseded::decode(bytes.as_slice()).unwrap();
+                kept.retain(|old| !withdraws(&s, old));
+            }
+            if let Some(bytes) = &entry.record {
+                let record = StreamRecord::decode(bytes.as_slice()).unwrap();
+                if matches!(record.item, Some(Item::Data(_))) {
+                    kept.push(record);
+                }
+            }
+        }
+        kept.into_iter()
+            .filter_map(|record| match record.item {
+                Some(Item::Data(data)) => Some(String::from_utf8(data.body.unwrap().data).unwrap()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    pub(crate) async fn batch(
+        &self,
+        producer: &Producer,
+        sequence: i64,
+        texts: &[&str],
+    ) -> StreamResult<Appended> {
+        self.write(producer, sequence, "", texts).await
+    }
+
+    async fn write(
+        &self,
+        producer: &Producer,
+        sequence: i64,
+        withdraw_run: &str,
+        texts: &[&str],
+    ) -> StreamResult<Appended> {
+        let records: Vec<_> = texts.iter().map(|text| record(producer, text)).collect();
+        let mut withdrawal = Withdrawal {
+            withdraw_run,
+            superseded: &[],
+            tasks: None,
+        };
+        let bytes = superseded(producer, &withdrawal);
+        withdrawal.superseded = &bytes;
+        self.store
+            .append(
+                &self.namespace,
+                &self.stream_ref(),
+                &Batch {
+                    producer,
+                    sequence,
+                    records: &records,
+                    digest: digest(texts),
+                    withdrawal,
+                },
+            )
+            .await
+    }
+
+    pub(crate) async fn close(&self, producer: &Producer) -> StreamResult<Closed> {
+        self.close_at(&self.stream_ref(), producer).await
+    }
+
+    /// Closes the Stream of `stream_ref`, with the owner fields it has.
+    pub(crate) async fn close_at(
+        &self,
+        stream_ref: &StreamRef,
+        producer: &Producer,
+    ) -> StreamResult<Closed> {
+        self.store
+            .close(&self.namespace, stream_ref, producer)
+            .await
+    }
+}
diff --git a/crates/streams/testdata/fields.json b/crates/streams/testdata/fields.json
index 93517dbde..8b9b269cd 100644
--- a/crates/streams/testdata/fields.json
+++ b/crates/streams/testdata/fields.json
@@ -2,9 +2,29 @@
   "kind": 3,
   "producer_id": "act:run%3A1:a:s",
   "run_id": "run:1",
-  "fence": "e|3|act:run%3A1:a:s",
-  "dedupe": "d|3|act:run%3A1:a:s",
-  "holds": "h|3|act:run%3A1:a:s",
+  "producer": "p|3|act:run%3A1:a:s",
   "run_wrote": "run|run%3A1",
-  "withdrawn_run": "wr|run%3A1"
+  "withdrawn_run": "wr|run%3A1",
+  "producer_values": [
+    {
+      "name": "a batch with Records",
+      "fence": "0000000000000000000000000000000000000000000000000000000068e778013b9ac9ff00000002",
+      "held": "0000000000000000000000000000000000000000000000000000000068e778013b9ac9ff00000002",
+      "sequence": 3,
+      "first": "1760000000123-4",
+      "last": "1760000000123-6",
+      "digest": "aca82bcf079b9b292bba1a09da05f527",
+      "value": "010000000000000000000000000000000000000000000000000000000068e778013b9ac9ff000000020000000000000000000000000000000000000000000000000000000068e778013b9ac9ff00000002000000000000000300000199c82cc07b0000000400000199c82cc07b00000006aca82bcf079b9b292bba1a09da05f527"
+    },
+    {
+      "name": "an empty batch that wrote no entry",
+      "fence": "0000000000000000000000000000000000000000000000000000000068e778020000000000000003",
+      "held": null,
+      "sequence": 1,
+      "first": "",
+      "last": "",
+      "digest": "e3b0c44298fc1c149afbf4c8996fb924",
+      "value": "000000000000000000000000000000000000000000000000000000000068e778020000000000000003000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000000000000000000000000000000e3b0c44298fc1c149afbf4c8996fb924"
+    }
+  ]
 }
diff --git a/crates/streams/testdata/gen_vectors.py b/crates/streams/testdata/gen_vectors.py
index 5e181a1bb..97ce51084 100644
--- a/crates/streams/testdata/gen_vectors.py
+++ b/crates/streams/testdata/gen_vectors.py
@@ -132,6 +132,45 @@ def epochs():
     return cases
 
 
+def entry_id(text):
+    """An entry id in a Producer field, the ms in 8 bytes and the seq in 4, zero for none."""
+    if not text:
+        return bytes(12)
+    ms, seq = (int(x) for x in text.split("-"))
+    return ms.to_bytes(8, "big") + seq.to_bytes(4, "big")
+
+
+def producer_value(fence, held, sequence, first, last, digest):
+    """The 129 bytes of a `p|` field."""
+    flags = bytes([1 if held else 0])
+    return (flags + fence + (held or bytes(40)) + sequence.to_bytes(8, "big") + entry_id(first)
+            + entry_id(last) + digest)
+
+
+def producer_values():
+    held = epoch_key({"attempt_scheduled_time": [1760000001, 999999999], "attempt": 2})
+    newer = epoch_key({"attempt_scheduled_time": [1760000002, 0], "attempt": 3})
+    cases = []
+    for name, fence, held_key, sequence, first, last, records in [
+        ("a batch with Records", held, held, 3, "1760000000123-4", "1760000000123-6",
+         [{"data": "01"}]),
+        ("an empty batch that wrote no entry", newer, None, 1, "", "", []),
+    ]:
+        batch_digest = bytes.fromhex(digest(records))
+        value = producer_value(fence, held_key, sequence, first, last, batch_digest)
+        cases.append({
+            "name": name,
+            "fence": fence.hex(),
+            "held": held_key.hex() if held_key else None,
+            "sequence": sequence,
+            "first": first,
+            "last": last,
+            "digest": batch_digest.hex(),
+            "value": value.hex(),
+        })
+    return cases
+
+
 def fields():
     run = "run:1"
     producer = "act:" + join(run, "a", "s")
@@ -139,11 +178,10 @@ def fields():
         "kind": 3,
         "producer_id": producer,
         "run_id": run,
-        "fence": f"e|3|{producer}",
-        "dedupe": f"d|3|{producer}",
-        "holds": f"h|3|{producer}",
+        "producer": f"p|3|{producer}",
         "run_wrote": f"run|{part(run)}",
         "withdrawn_run": f"wr|{part(run)}",
+        "producer_values": producer_values(),
     }
 
 
diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs
index 853a4e96c..d824561a4 100644
--- a/crates/streams/tests/vectors.rs
+++ b/crates/streams/tests/vectors.rs
@@ -12,9 +12,10 @@ use temporalio_streams::{
     ids::{self, StreamKeys},
     offset::{entry_id_of_offset, offset_of_entry_id},
     proto::{
-        ActivityEpoch, AppendRecord, ClientEpoch, LocalActivityEpoch, Producer, ProducerEpoch,
-        StreamData, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch,
-        producer_epoch::Kind, stream_record::Item,
+        ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer,
+        LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, StreamData,
+        StreamRecord, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch,
+        producer::Kind, stream_record::Item,
     },
 };
 
@@ -39,31 +40,43 @@ fn number(value: &Value, field: &str) -> i64 {
     value[field].as_i64().unwrap_or(0)
 }
 
-/// The epoch of a kind, from the flat fields of the vectors. The case takes the fields of its kind.
-fn epoch(kind: &str, value: &Value) -> ProducerEpoch {
+/// The Producer kind of a vector, from its flat fields and its run. The case takes the fields of
+/// its kind, and a run-based kind carries the run. `empty` has no case.
+fn kind_of(kind: &str, value: &Value, run_id: &str) -> Option<Kind> {
     let task = WorkflowTaskEpoch {
         started_event_id: number(value, "task_started_event_id"),
         attempt: number(value, "task_attempt") as i32,
         started_time: time(&value["task_started_time"]),
     };
     let attempt = number(value, "attempt") as i32;
-    let activity = ActivityEpoch {
-        attempt_scheduled_time: time(&value["attempt_scheduled_time"]),
-        attempt,
-    };
-    let kind = match kind {
-        "workflow" => Some(Kind::Workflow(task)),
-        "local_activity" => Some(Kind::LocalActivity(LocalActivityEpoch {
-            task: Some(task),
+    let run_id = run_id.to_string();
+    let activity = ActivityProducer {
+        run_id: run_id.clone(),
+        epoch: Some(ActivityEpoch {
+            attempt_scheduled_time: time(&value["attempt_scheduled_time"]),
             attempt,
+        }),
+    };
+    match kind {
+        "workflow" => Some(Kind::Workflow(WorkflowProducer {
+            run_id,
+            epoch: Some(task),
+        })),
+        "local_activity" => Some(Kind::LocalActivity(LocalActivityProducer {
+            run_id,
+            epoch: Some(LocalActivityEpoch {
+                task: Some(task),
+                attempt,
+            }),
         })),
         "activity" => Some(Kind::Activity(activity)),
         "standalone_activity" => Some(Kind::StandaloneActivity(activity)),
-        "client" => Some(Kind::Client(ClientEpoch { attempt })),
+        "client" => Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt }),
+        })),
         "empty" => None,
         other => panic!("unknown kind {other}"),
-    };
-    ProducerEpoch { kind }
+    }
 }
 
 #[test]
@@ -137,10 +150,10 @@ fn offsets_match() {
 #[test]
 fn epoch_keys_match() {
     for case in load("epochs.json").as_array().unwrap() {
-        let epoch = epoch(text(case, "kind"), &case["epoch"]);
-        assert_eq!(epoch_key(&epoch).to_vec(), hex(text(case, "key")));
+        let kind = kind_of(text(case, "kind"), &case["epoch"], "");
+        assert_eq!(epoch_key(kind.as_ref()).to_vec(), hex(text(case, "key")));
         assert_eq!(
-            task_key(&task_of(&epoch)).to_vec(),
+            task_key(&task_of(kind.as_ref())).to_vec(),
             hex(text(case, "task_key"))
         );
     }
@@ -152,29 +165,34 @@ fn the_appendix_e_withdrawals_match() {
     assert!(cases.as_array().unwrap().len() >= 35);
     for case in cases.as_array().unwrap() {
         let s = &case["superseded"];
+        // A create's item has no epoch, and a Reader must not read it as an empty one.
         let withdrawing = Producer {
             id: text(s, "producer_id").to_string(),
-            // A create's item has no epoch, and a Reader must not read it as an empty one.
-            epoch: s
-                .get("epoch")
-                .map(|value| epoch(text(s, "producer_kind"), value)),
-            run_id: s["run_id"].as_str().unwrap_or_default().to_string(),
+            kind: s.get("epoch").and_then(|value| {
+                kind_of(
+                    text(s, "producer_kind"),
+                    value,
+                    s["run_id"].as_str().unwrap_or_default(),
+                )
+            }),
         };
         let superseded = StreamSuperseded {
             writer: Some(StreamWriter {
                 producer: Some(withdrawing),
-                previous_completed_task: s.get("previous_completed_task").map(|c| {
-                    WorkflowTaskEpoch {
-                        started_event_id: number(c, "started_event_id"),
-                        attempt: number(c, "attempt") as i32,
-                        started_time: time(&c["started_time"]),
-                    }
+                withdrawal: Some(RunWithdrawal {
+                    previous_completed_task: s.get("previous_completed_task").map(|c| {
+                        WorkflowTaskEpoch {
+                            started_event_id: number(c, "started_event_id"),
+                            attempt: number(c, "attempt") as i32,
+                            started_time: time(&c["started_time"]),
+                        }
+                    }),
+                    failed_tasks: s["failed_tasks"]
+                        .as_array()
+                        .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect())
+                        .unwrap_or_default(),
+                    withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(),
                 }),
-                failed_tasks: s["failed_tasks"]
-                    .as_array()
-                    .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect())
-                    .unwrap_or_default(),
-                withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(),
             }),
         };
         let p = &case["record"]["producer"];
@@ -182,8 +200,13 @@ fn the_appendix_e_withdrawals_match() {
             item: Some(Item::Data(StreamData {
                 producer: Some(Producer {
                     id: text(p, "id").to_string(),
-                    epoch: p.get("epoch").map(|value| epoch(text(p, "kind"), value)),
-                    run_id: p["run_id"].as_str().unwrap_or_default().to_string(),
+                    kind: p.get("epoch").and_then(|value| {
+                        kind_of(
+                            text(p, "kind"),
+                            value,
+                            p["run_id"].as_str().unwrap_or_default(),
+                        )
+                    }),
                 }),
                 ..Default::default()
             })),