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 fcec138bd..f9fb65566 100644
--- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
+++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto
@@ -33,11 +33,12 @@ message MemoryStreamConfig {
 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.
+    // A write sends at most this many Records in one batch. Zero, or a value above 1000, is 1000.
     int32 max_batch_size = 4;
     // How long a write retries a Store failure. Empty is 10 s.
     google.protobuf.Duration max_retry_duration = 5;
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 bc3ffb8af..6fd9a271e 100644
--- a/crates/streams/src/lib.rs
+++ b/crates/streams/src/lib.rs
@@ -19,6 +19,7 @@ mod service;
 pub mod store;
 #[cfg(any(test, feature = "test-support"))]
 pub mod testing;
+mod writer;
 
 pub use error::{Reason, StreamError, StreamResult};
 pub use runtime::{MemoryStreamRuntime, StoreFactory, StreamRuntime};
diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs
index 589334a59..de076e762 100644
--- a/crates/streams/src/memory.rs
+++ b/crates/streams/src/memory.rs
@@ -5,7 +5,9 @@
 use crate::{
     Reason, StreamError, StreamResult,
     digest::DIGEST_LEN,
-    epoch::{EPOCH_KEY_LEN, ProducerKind, TASK_KEY_LEN, epoch_key, task_key, task_of},
+    epoch::{
+        EPOCH_KEY_LEN, ProducerKind, TASK_KEY_LEN, epoch_key, producer_run_id, task_key, task_of,
+    },
     ids::{StreamKeys, encode_part},
     offset::{entry_id_of_offset, offset_of_entry_id},
     proto::{Producer, StreamRef, WorkflowTaskEpoch},
@@ -693,8 +695,8 @@ impl Store for MemoryStore {
         let stream_id = stream_ref.id.as_str();
         let producer = batch.producer;
         let kind = ProducerKind::try_from(producer)?;
-        let epoch = epoch_key(&producer.epoch.unwrap_or_default());
-        let run = encode_run(&producer.run_id);
+        let epoch = epoch_key(producer.kind.as_ref());
+        let run = encode_run(producer_run_id(producer));
         let withdrawal = &batch.withdrawal;
         let withdrawn_run = encode_run(withdrawal.withdraw_run);
         let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE));
@@ -702,7 +704,7 @@ impl Store for MemoryStore {
         let member = format!("{}|{}", kind as i32, producer.id);
         // The Workflow Task of a Local Activity of a run, which the run's task facts can withdraw.
         let task = (kind == ProducerKind::LocalActivity && !run.is_empty())
-            .then(|| task_key(&task_of(&producer.epoch.unwrap_or_default())));
+            .then(|| task_key(&task_of(producer.kind.as_ref())));
         let facts = withdrawal.tasks.filter(|_| task.is_some());
         let completed = facts.and_then(|facts| facts.completed).map(task_key);
         let open_field = format!("oa|{run}");
@@ -944,8 +946,8 @@ impl Store for MemoryStore {
     ) -> StreamResult<Closed> {
         let stream_id = stream_ref.id.as_str();
         let kind = ProducerKind::try_from(producer)?;
-        let epoch = epoch_key(&producer.epoch.unwrap_or_default());
-        let run = encode_run(&producer.run_id);
+        let epoch = epoch_key(producer.kind.as_ref());
+        let run = encode_run(producer_run_id(producer));
         let field = format!("p|{}|{}", kind as i32, producer.id);
         let tombstone = millis(TOMBSTONE);
         let chain = owner_chain(stream_ref);
diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs
index 85d32e1f4..2277adf9b 100644
--- a/crates/streams/src/record.rs
+++ b/crates/streams/src/record.rs
@@ -3,8 +3,8 @@
 use crate::{
     Reason, StreamError, StreamResult,
     proto::{
-        AppendRecord, Producer, StreamData, StreamRecord, StreamSuperseded, StreamWriter,
-        stream_record::Item,
+        AppendRecord, Producer, RunWithdrawal, StreamData, StreamRecord, StreamSuperseded,
+        StreamWriter, stream_record::Item,
     },
     store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES},
 };
@@ -65,11 +65,14 @@ pub(crate) fn data_records(
 
 /// The facts a `Superseded` item of this batch gives the Reader.
 pub(crate) fn superseded(producer: &Producer, withdraw_run: &str) -> Vec<u8> {
+    let withdrawal = (!withdraw_run.is_empty()).then(|| RunWithdrawal {
+        withdraw_run: withdraw_run.to_string(),
+        ..Default::default()
+    });
     StreamSuperseded {
         writer: Some(StreamWriter {
             producer: Some(producer.clone()),
-            withdraw_run: withdraw_run.to_string(),
-            ..Default::default()
+            withdrawal,
         }),
     }
     .encode_to_vec()
diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs
index 6c54a461d..9f21f687b 100644
--- a/crates/streams/src/runtime.rs
+++ b/crates/streams/src/runtime.rs
@@ -5,11 +5,14 @@ use crate::{
     memory::MemoryStore,
     metrics::Metrics,
     proto::{
-        AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, MemoryStreamConfig,
-        ReadRecordsRequest,
+        AppendRecordsRequest, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse,
+        CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, FreeWriterRequest,
+        FreeWriterResponse, MemoryStreamConfig, OpenWriterRequest, OpenWriterResponse,
+        ReadRecordsRequest, WriteRecordsRequest, WriteRecordsResponse,
     },
     service::StreamService,
     store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store},
+    writer::Writers,
 };
 use prost::Message;
 use std::{sync::Arc, time::Duration};
@@ -44,6 +47,7 @@ pub(crate) struct Shared {
     metrics: Metrics,
     factory: Arc<dyn StoreFactory>,
     service: OnceCell<Arc<StreamService>>,
+    writers: Writers,
 }
 
 impl Shared {
@@ -97,6 +101,7 @@ impl StreamRuntime {
             metrics: Metrics::new(&meter),
             factory,
             service: OnceCell::new(),
+            writers: Writers::default(),
         });
         Arc::new(Self { shared })
     }
@@ -116,9 +121,21 @@ impl StreamRuntime {
         self.shared.name
     }
 
+    /// Releases one reference to a writer with no wait, as the `FreeWriter` method does. A bridge
+    /// calls it from a finalizer, which has no async runtime to await on.
+    pub fn free_writer(&self, writer_id: u64) -> StreamResult<()> {
+        self.shared.writers.free(writer_id)
+    }
+
     /// Runs one method on encoded input and returns the encoded output. The bridge raises an error
     /// with [StreamError::code] and [StreamError::details].
     pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult<Vec<u8>> {
+        // `FreeWriter` needs no Store, so it works when the Store is down too.
+        if method == "FreeWriter" {
+            let request = decode::<FreeWriterRequest>(input)?;
+            self.free_writer(request.writer_id)?;
+            return Ok(FreeWriterResponse {}.encode_to_vec());
+        }
         let service = self.shared.service().await?;
         let output = match method {
             "CreateStream" => service.create_stream(decode(input)?).await?.encode_to_vec(),
@@ -138,6 +155,37 @@ impl StreamRuntime {
                 .describe_stream(decode::<DescribeStreamRequest>(input)?)
                 .await?
                 .encode_to_vec(),
+            "OpenWriter" => {
+                let request = decode::<OpenWriterRequest>(input)?;
+                let writer_id = self.shared.writers.open(service, request).await?;
+                OpenWriterResponse { writer_id }.encode_to_vec()
+            }
+            "WriteRecords" => {
+                let request = decode::<WriteRecordsRequest>(input)?;
+                let (first_offset, last_offset) = self
+                    .shared
+                    .writers
+                    .write(request.writer_id, request.records)
+                    .await?;
+                WriteRecordsResponse {
+                    first_offset,
+                    last_offset,
+                }
+                .encode_to_vec()
+            }
+            "CloseWriters" => {
+                decode::<CloseWritersRequest>(input)?;
+                let failed = self.shared.writers.close_all().await;
+                CloseWritersResponse {
+                    failed: failed as i32,
+                }
+                .encode_to_vec()
+            }
+            "CloseWriter" => {
+                let request = decode::<CloseWriterRequest>(input)?;
+                let last_offset = self.shared.writers.close(request.writer_id).await?;
+                CloseWriterResponse { last_offset }.encode_to_vec()
+            }
             _ => {
                 return Err(StreamError::new(
                     Reason::Unsupported,
@@ -200,12 +248,17 @@ pub(crate) fn decode<M: Message + Default>(input: &[u8]) -> StreamResult<M> {
 mod tests {
     use super::*;
 
-    const METHODS: [&str; 5] = [
+    const METHODS: [&str; 10] = [
         "CreateStream",
         "AppendRecords",
         "CloseStream",
         "ReadRecords",
         "DescribeStream",
+        "OpenWriter",
+        "WriteRecords",
+        "CloseWriters",
+        "CloseWriter",
+        "FreeWriter",
     ];
 
     fn runtime() -> Arc<StreamRuntime> {
diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs
index b617e00e4..b074cbe53 100644
--- a/crates/streams/src/service.rs
+++ b/crates/streams/src/service.rs
@@ -4,15 +4,18 @@
 use crate::{
     Reason, StreamError, StreamResult,
     digest::batch_digest,
-    epoch::{ProducerKind, task_of, withdraws},
+    epoch::{
+        ProducerKind, producer_run_id, task_of, withdraws, writer_failed_tasks,
+        writer_previous_completed_task, writer_withdraw_run,
+    },
     ids::{check_producer, check_stream_id},
     metrics::Metrics,
     proto::{
         AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse,
         CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse,
-        Producer, ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamIdConflictPolicy,
+        Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, StreamIdConflictPolicy,
         StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter,
-        WorkflowTaskEpoch, producer_epoch::Kind, stream_record::Item,
+        WorkflowProducer, WorkflowTaskEpoch, producer::Kind, stream_record::Item,
     },
     record::{data_records, superseded},
     store::{
@@ -108,7 +111,7 @@ fn stream_id(stream_ref: Option<&StreamRef>) -> StreamResult<&str> {
 /// so a Reader withdraws only that run's Records.
 fn run_superseded(owner: Option<&Producer>, withdraw_run: &str) -> Vec<u8> {
     let producer = Producer {
-        epoch: None,
+        kind: None,
         ..owner.cloned().unwrap_or_default()
     };
     superseded(&producer, withdraw_run)
@@ -121,20 +124,23 @@ fn orphan_superseded(
     completed: Option<&WorkflowTaskEpoch>,
     failed_tasks: &[i64],
 ) -> Vec<u8> {
-    let task = producer.epoch.as_ref().map(task_of).unwrap_or_default();
+    let run_id = producer_run_id(producer);
+    let task = task_of(producer.kind.as_ref());
     let workflow = Producer {
-        id: crate::ids::workflow_producer_id(&producer.run_id),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Workflow(task)),
-        }),
-        run_id: producer.run_id.clone(),
+        id: crate::ids::workflow_producer_id(run_id),
+        kind: Some(Kind::Workflow(WorkflowProducer {
+            run_id: run_id.to_string(),
+            epoch: Some(task),
+        })),
     };
     StreamSuperseded {
         writer: Some(StreamWriter {
             producer: Some(workflow),
-            previous_completed_task: completed.copied(),
-            failed_tasks: failed_tasks.to_vec(),
-            ..Default::default()
+            withdrawal: Some(RunWithdrawal {
+                previous_completed_task: completed.copied(),
+                failed_tasks: failed_tasks.to_vec(),
+                withdraw_run: String::new(),
+            }),
         }),
     }
     .encode_to_vec()
@@ -151,9 +157,7 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> {
         ));
     }
     // Without its task, a re-dispatch in a later task has the same epoch as the first dispatch.
-    let task = producer
-        .epoch
-        .map_or(0, |epoch| task_of(&epoch).started_event_id);
+    let task = task_of(producer.kind.as_ref()).started_event_id;
     if kind == ProducerKind::LocalActivity && task <= 0 {
         return Err(StreamError::invalid(
             "A Local Activity Producer's epoch must name the Workflow Task that dispatched it",
@@ -165,8 +169,8 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> {
 /// Refuses task facts that a Store would drop or misread. Only a Local Activity has its run's task
 /// facts, and a failed task is a started event id, which is never zero or negative.
 fn check_task_facts(kind: ProducerKind, writer: &StreamWriter) -> StreamResult<()> {
-    let completed = writer.previous_completed_task.as_ref();
-    if completed.is_none() && writer.failed_tasks.is_empty() {
+    let completed = writer_previous_completed_task(writer);
+    if completed.is_none() && writer_failed_tasks(writer).is_empty() {
         return Ok(());
     }
     if kind != ProducerKind::LocalActivity {
@@ -180,7 +184,7 @@ fn check_task_facts(kind: ProducerKind, writer: &StreamWriter) -> StreamResult<(
         ));
     }
     if completed.is_some_and(|c| c.started_event_id < 0)
-        || writer.failed_tasks.iter().any(|&id| id <= 0)
+        || writer_failed_tasks(writer).iter().any(|&id| id <= 0)
     {
         return Err(StreamError::invalid(
             "A task's started event id must be positive",
@@ -243,7 +247,7 @@ impl StreamService {
                     "A create with withdraw_run needs the owner run",
                 ));
             };
-            if request.withdraw_run == owner.run_id {
+            if request.withdraw_run == producer_run_id(owner) {
                 return Err(StreamError::invalid("A run doesn't withdraw itself"));
             }
             self.store.capabilities().require(Capabilities::RUN_FENCE)?;
@@ -302,10 +306,11 @@ impl StreamService {
                 .capabilities()
                 .require(Capabilities::NEXT_SEQUENCE)?;
         }
-        if !writer.withdraw_run.is_empty() {
+        let withdraw_run = writer_withdraw_run(&writer);
+        if !withdraw_run.is_empty() {
             // Only a retry run's Local Activity withdraws its previous run, as only the owner's
             // create does. Any other Producer could withdraw and fence an unrelated run.
-            if kind != ProducerKind::LocalActivity || writer.withdraw_run == producer.run_id {
+            if kind != ProducerKind::LocalActivity || withdraw_run == producer_run_id(producer) {
                 return Err(StreamError::invalid(
                     "Only a Local Activity of a later run withdraws a run",
                 ));
@@ -315,26 +320,24 @@ impl StreamService {
         check_task_facts(kind, &writer)?;
         let digest = batch_digest(&request.records);
         let records = data_records(producer, request.records)?;
-        let bytes = superseded(producer, &writer.withdraw_run);
+        let bytes = superseded(producer, withdraw_run);
         // A Local Activity sends its run's task facts, so that the batch withdraws the run's
         // orphaned Local Activity Records. These are the Records of the tasks that never completed.
-        let completed = writer
-            .previous_completed_task
-            .filter(|c| c.started_event_id > 0);
+        let completed = writer_previous_completed_task(&writer).filter(|c| c.started_event_id > 0);
         let orphans = (kind == ProducerKind::LocalActivity
-            && writer.previous_completed_task.is_some())
-        .then(|| orphan_superseded(producer, completed.as_ref(), &writer.failed_tasks));
+            && writer_previous_completed_task(&writer).is_some())
+        .then(|| orphan_superseded(producer, completed, writer_failed_tasks(&writer)));
         let batch = Batch {
             producer,
             sequence: request.sequence,
             records: &records,
             digest,
             withdrawal: Withdrawal {
-                withdraw_run: &writer.withdraw_run,
+                withdraw_run,
                 superseded: &bytes,
                 tasks: orphans.as_ref().map(|orphans| TaskFacts {
-                    completed: completed.as_ref(),
-                    failed_tasks: &writer.failed_tasks,
+                    completed,
+                    failed_tasks: writer_failed_tasks(&writer),
                     superseded: orphans,
                 }),
             },
diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs
index 350ab7964..94b3b5c85 100644
--- a/crates/streams/src/testing/mod.rs
+++ b/crates/streams/src/testing/mod.rs
@@ -6,6 +6,7 @@ mod support;
 pub mod fields;
 pub mod service;
 pub mod store;
+pub mod writer;
 
 pub use support::Needs;
 
@@ -153,4 +154,9 @@ mod tests {
     async fn the_service_suite_passes_on_the_memory_store() {
         run(vec![Arc::new(MemoryBackend)], service::CASES).await;
     }
+
+    #[tokio::test]
+    async fn the_writer_suite_passes_on_the_memory_store() {
+        run(vec![Arc::new(MemoryBackend)], writer::CASES).await;
+    }
 }
diff --git a/crates/streams/src/testing/service.rs b/crates/streams/src/testing/service.rs
index 11191fe87..7cc54d9fe 100644
--- a/crates/streams/src/testing/service.rs
+++ b/crates/streams/src/testing/service.rs
@@ -6,15 +6,16 @@ use super::{case, support::Needs};
 
 use crate::{
     StreamError, StreamRuntime,
-    epoch::withdraws,
+    epoch::{withdraws, writer_withdraw_run},
     ids::{activity_producer_id, client_producer_id, named_client_producer_id},
     proto::{
-        ActivityEpoch, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch,
-        CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse,
-        DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, Producer, ProducerEpoch,
-        ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamIdConflictPolicy,
-        StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter,
-        WorkflowTaskEpoch, producer_epoch::Kind, stream_record::Item,
+        ActivityEpoch, ActivityProducer, AppendRecord, AppendRecordsRequest, AppendRecordsResponse,
+        ClientEpoch, ClientProducer, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest,
+        CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch,
+        LocalActivityProducer, Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal,
+        StreamErrorDetails, StreamIdConflictPolicy, StreamReadStart, StreamRecord, StreamRef,
+        StreamStatus, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch,
+        producer::Kind, stream_record::Item,
     },
 };
 use prost::Message;
@@ -88,10 +89,9 @@ fn writer_for(producer: Producer) -> Option<StreamWriter> {
 fn client(attempt: i32) -> Producer {
     Producer {
         id: client_producer_id("me", "p"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Client(ClientEpoch { attempt })),
-        }),
-        run_id: String::new(),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt }),
+        })),
     }
 }
 
@@ -108,16 +108,16 @@ fn named(name: &str, attempt: i32) -> Producer {
 fn activity(attempt: i32, scheduled: i64) -> Producer {
     Producer {
         id: activity_producer_id("r1", "a", "s"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Activity(ActivityEpoch {
+        kind: Some(Kind::Activity(ActivityProducer {
+            run_id: "r1".to_string(),
+            epoch: Some(ActivityEpoch {
                 attempt_scheduled_time: Some(Timestamp {
                     seconds: scheduled,
                     nanos: 0,
                 }),
                 attempt,
-            })),
-        }),
-        run_id: "r1".to_string(),
+            }),
+        })),
     }
 }
 
@@ -125,10 +125,10 @@ fn activity(attempt: i32, scheduled: i64) -> Producer {
 fn workflow(run: &str) -> Producer {
     Producer {
         id: crate::ids::workflow_producer_id(run),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())),
-        }),
-        run_id: run.to_string(),
+        kind: Some(Kind::Workflow(WorkflowProducer {
+            run_id: run.to_string(),
+            epoch: Some(WorkflowTaskEpoch::default()),
+        })),
     }
 }
 
@@ -388,13 +388,13 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() {
         let run = format!("{}-run", case.stream);
         let activity = Producer {
             id: activity_producer_id(&run, "a", "s"),
-            epoch: Some(ProducerEpoch {
-                kind: Some(Kind::Activity(ActivityEpoch {
+            kind: Some(Kind::Activity(ActivityProducer {
+                run_id: run.clone(),
+                epoch: Some(ActivityEpoch {
                     attempt: 1,
                     ..Default::default()
-                })),
-            }),
-            run_id: run.clone(),
+                }),
+            })),
         };
         let error = case.close(activity).await.unwrap_err();
         assert_eq!(
@@ -432,13 +432,13 @@ async fn a_create_checks_its_owner_and_its_ref() {
         };
         let local_activity = Producer {
             id: crate::ids::local_activity_producer_id(&run, "call-1"),
-            epoch: Some(ProducerEpoch {
-                kind: Some(Kind::LocalActivity(LocalActivityEpoch {
+            kind: Some(Kind::LocalActivity(LocalActivityProducer {
+                run_id: run.clone(),
+                epoch: Some(LocalActivityEpoch {
                     task: Some(task),
                     attempt: 1,
-                })),
-            }),
-            ..workflow.clone()
+                }),
+            })),
         };
         let chain = |workflow_id: &str| StreamRef {
             id: case.stream.clone(),
@@ -500,11 +500,12 @@ async fn a_create_checks_its_owner_and_its_ref() {
 async fn a_producer_must_match_its_kind_and_not_be_a_workflow() {
     for case in cases_for(Needs::CALLS).await {
         case.create(None).await.unwrap();
-        let activity = ProducerEpoch {
-            kind: Some(Kind::Activity(ActivityEpoch::default())),
-        };
-        for epoch in [Some(activity), None, Some(ProducerEpoch::default())] {
-            let wrong = Producer { epoch, ..client(1) };
+        let activity = Kind::Activity(ActivityProducer {
+            run_id: String::new(),
+            epoch: Some(ActivityEpoch::default()),
+        });
+        for kind in [Some(activity), None] {
+            let wrong = Producer { kind, ..client(1) };
             let error = case.append(wrong, 1, vec![record("x")]).await.unwrap_err();
             assert_eq!(
                 (error.code(), reason(&error).as_str()),
@@ -515,13 +516,13 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() {
         assert_eq!(error.code(), Code::InvalidArgument);
         let no_task = Producer {
             id: crate::ids::local_activity_producer_id("r1", "call-1"),
-            run_id: "r1".to_string(),
-            epoch: Some(ProducerEpoch {
-                kind: Some(Kind::LocalActivity(LocalActivityEpoch {
+            kind: Some(Kind::LocalActivity(LocalActivityProducer {
+                run_id: "r1".to_string(),
+                epoch: Some(LocalActivityEpoch {
                     task: None,
                     attempt: 1,
-                })),
-            }),
+                }),
+            })),
         };
         let error = case
             .append(no_task, 1, vec![record("x")])
@@ -1044,8 +1045,10 @@ async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() {
             r#ref: case.stream_ref(),
             writer: Some(StreamWriter {
                 producer: Some(local_activity(run)),
-                withdraw_run: withdraw_run.to_string(),
-                ..Default::default()
+                withdrawal: Some(RunWithdrawal {
+                    withdraw_run: withdraw_run.to_string(),
+                    ..Default::default()
+                }),
             }),
             sequence: 1,
             records: vec![record(text)],
@@ -1080,11 +1083,12 @@ async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() {
         assert_eq!(texts(&read), ["old", "<superseded>", "new"]);
         let superseded = superseded_of(&read.records[1]);
         let writer = superseded.writer.as_ref().unwrap();
-        assert_eq!(writer.withdraw_run, previous);
-        // The item's writer has its Producer, which the WIT requires.
+        assert_eq!(writer_withdraw_run(writer), previous);
+        // The item's writer has its Producer, which the WIT requires. A create's item has no kind,
+        // so a Reader can't read it as an epoch, and the run stays in the id.
         let producer = writer.producer.as_ref().unwrap();
         assert_eq!(producer.id, crate::ids::workflow_producer_id(&retry));
-        assert_eq!(producer.run_id, retry);
+        assert!(producer.kind.is_none());
         let error = call::<_, CreateStreamResponse>(
             &case.ext,
             "CreateStream",
@@ -1124,8 +1128,10 @@ async fn only_a_later_run_s_local_activity_withdraws_a_run_on_append() {
                     r#ref: case.stream_ref(),
                     writer: Some(StreamWriter {
                         producer: Some(producer),
-                        withdraw_run: run.clone(),
-                        ..Default::default()
+                        withdrawal: Some(RunWithdrawal {
+                            withdraw_run: run.clone(),
+                            ..Default::default()
+                        }),
                     }),
                     sequence: 1,
                     records: vec![record("other")],
@@ -1153,13 +1159,13 @@ fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer {
     };
     Producer {
         id: crate::ids::local_activity_producer_id(run, "call-1"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::LocalActivity(LocalActivityEpoch {
+        kind: Some(Kind::LocalActivity(LocalActivityProducer {
+            run_id: run.to_string(),
+            epoch: Some(LocalActivityEpoch {
                 task: Some(task),
                 attempt: 1,
-            })),
-        }),
-        run_id: run.to_string(),
+            }),
+        })),
     }
 }
 
@@ -1242,11 +1248,13 @@ fn local_activity_write(
         r#ref: case.stream_ref(),
         writer: Some(StreamWriter {
             producer: Some(producer),
-            previous_completed_task: Some(WorkflowTaskEpoch {
-                started_event_id: completed,
+            withdrawal: Some(RunWithdrawal {
+                previous_completed_task: Some(WorkflowTaskEpoch {
+                    started_event_id: completed,
+                    ..Default::default()
+                }),
                 ..Default::default()
             }),
-            ..Default::default()
         }),
         sequence: 1,
         records: vec![record(text)],
@@ -1254,11 +1262,18 @@ fn local_activity_write(
     }
 }
 
-/// The writer of a write, to change its Producer or its task facts.
+/// The writer of a write, to change its Producer.
 fn writer_mut(request: &mut AppendRecordsRequest) -> &mut StreamWriter {
     request.writer.get_or_insert_with(Default::default)
 }
 
+/// The withdrawal facts of a write, to change its task facts.
+fn withdrawal_mut(request: &mut AppendRecordsRequest) -> &mut RunWithdrawal {
+    writer_mut(request)
+        .withdrawal
+        .get_or_insert_with(Default::default)
+}
+
 /// The close Activity closes a Workflow's Stream as the run's Workflow Producer, which the service
 /// refuses on an append.
 async fn the_close_activity_closes_a_workflow_stream_as_its_workflow_producer() {
@@ -1374,7 +1389,7 @@ async fn a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id() {
         .unwrap();
         // Task [3] failed and task [4] completed with no write here. Task [5] knows both.
         let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next");
-        writer_mut(&mut next).failed_tasks = vec![3];
+        withdrawal_mut(&mut next).failed_tasks = vec![3];
         let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap();
         let read = case.read(ReadRecordsRequest::default()).await.unwrap();
         assert_eq!(texts(&read), ["orphan", "<superseded>", "next"]);
@@ -1399,7 +1414,7 @@ async fn a_first_write_of_a_task_that_never_completed_is_fenced() {
         .unwrap();
         // Task [3] runs call-2 and fails before call-2 writes. Task [4] knows that [3] failed.
         let mut retry = local_activity_write(&case, &run, "call-3", 4, 2, "retry");
-        writer_mut(&mut retry).failed_tasks = vec![3];
+        withdrawal_mut(&mut retry).failed_tasks = vec![3];
         let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", retry).await.unwrap();
         // Call-1 of the completed task [2] still runs and writes with its older facts.
         let more = AppendRecordsRequest {
@@ -1428,11 +1443,11 @@ async fn a_write_with_bad_task_facts_fails() {
         case.create(None).await.unwrap();
         let facts = |failed: Vec<i64>| {
             let mut request = local_activity_write(&case, &run, "call-1", 4, 2, "bad");
-            writer_mut(&mut request).failed_tasks = failed;
+            withdrawal_mut(&mut request).failed_tasks = failed;
             request
         };
         let mut no_completed = facts(vec![3]);
-        writer_mut(&mut no_completed).previous_completed_task = None;
+        withdrawal_mut(&mut no_completed).previous_completed_task = None;
         let mut by_client = facts(vec![]);
         writer_mut(&mut by_client).producer = Some(client(1));
         let mut by_activity = facts(vec![3]);
@@ -1461,10 +1476,10 @@ async fn a_write_with_no_task_facts_is_withdrawn_by_a_later_write() {
         let run = format!("{}-run", case.stream);
         case.create(None).await.unwrap();
         let mut orphan = local_activity_write(&case, &run, "call-1", 3, 2, "orphan");
-        writer_mut(&mut orphan).previous_completed_task = None;
+        withdrawal_mut(&mut orphan).previous_completed_task = None;
         let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", orphan).await.unwrap();
         let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next");
-        writer_mut(&mut next).failed_tasks = vec![3];
+        withdrawal_mut(&mut next).failed_tasks = vec![3];
         let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap();
         let read = case.read(ReadRecordsRequest::default()).await.unwrap();
         assert_eq!(texts(&read), ["orphan", "<superseded>", "next"]);
@@ -1488,7 +1503,7 @@ async fn a_batch_writes_an_orphan_item_and_an_epoch_item() {
         .await
         .unwrap();
         let mut rerun = local_activity_write(&case, &run, "call-1", 5, 4, "rerun");
-        writer_mut(&mut rerun).failed_tasks = vec![3];
+        withdrawal_mut(&mut rerun).failed_tasks = vec![3];
         let answer: AppendRecordsResponse = call(&case.ext, "AppendRecords", rerun).await.unwrap();
         let read = case.read(ReadRecordsRequest::default()).await.unwrap();
         assert_eq!(
@@ -1499,9 +1514,16 @@ async fn a_batch_writes_an_orphan_item_and_an_epoch_item() {
             superseded_of(&read.records[1]),
             superseded_of(&read.records[2]),
         );
-        let kind = |item: &StreamSuperseded| {
-            let producer = item.writer.as_ref()?.producer.as_ref()?;
-            producer.epoch?.kind
+        let kind = |item: &StreamSuperseded| -> Option<Kind> {
+            Some(
+                item.writer
+                    .as_ref()?
+                    .producer
+                    .as_ref()?
+                    .kind
+                    .as_ref()?
+                    .clone(),
+            )
         };
         assert!(matches!(kind(orphans), Some(Kind::Workflow(_))));
         assert!(matches!(kind(epoch), Some(Kind::LocalActivity(_))));
@@ -1569,21 +1591,23 @@ fn task_write(
 ) -> AppendRecordsRequest {
     let producer = Producer {
         id: crate::ids::local_activity_producer_id(run, call),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::LocalActivity(LocalActivityEpoch {
+        kind: Some(Kind::LocalActivity(LocalActivityProducer {
+            run_id: run.to_string(),
+            epoch: Some(LocalActivityEpoch {
                 task: Some(task),
                 attempt: 1,
-            })),
-        }),
-        run_id: run.to_string(),
+            }),
+        })),
     };
     AppendRecordsRequest {
         namespace: case.namespace.clone(),
         r#ref: case.stream_ref(),
         writer: Some(StreamWriter {
             producer: Some(producer),
-            previous_completed_task: completed,
-            ..Default::default()
+            withdrawal: Some(RunWithdrawal {
+                previous_completed_task: completed,
+                ..Default::default()
+            }),
         }),
         sequence: 1,
         records: vec![record(text)],
diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs
index 5715a42f2..9221007d4 100644
--- a/crates/streams/src/testing/store.rs
+++ b/crates/streams/src/testing/store.rs
@@ -18,8 +18,8 @@ use crate::{
     memory::{EntryId, ProducerField},
     offset::{entry_id_of_offset, offset_of_entry_id},
     proto::{
-        ActivityEpoch, ClientEpoch, Producer, ProducerEpoch, StreamRef, WorkflowTaskEpoch,
-        producer_epoch::Kind,
+        ActivityEpoch, ActivityProducer, ClientEpoch, ClientProducer, Producer, StreamRef,
+        WorkflowProducer, WorkflowTaskEpoch, producer::Kind,
     },
     store::{Batch, ConflictPolicy, Create, CreateOutcome, Created, Owner, TaskFacts, Withdrawal},
 };
@@ -30,26 +30,25 @@ use std::time::Duration;
 fn activity(run: &str, attempt: i32, scheduled: i64) -> Producer {
     Producer {
         id: activity_producer_id(run, "a", "s"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Activity(ActivityEpoch {
+        kind: Some(Kind::Activity(ActivityProducer {
+            run_id: run.to_string(),
+            epoch: Some(ActivityEpoch {
                 attempt_scheduled_time: Some(Timestamp {
                     seconds: scheduled,
                     nanos: 0,
                 }),
                 attempt,
-            })),
-        }),
-        run_id: run.to_string(),
+            }),
+        })),
     }
 }
 
 fn client(attempt: i32) -> Producer {
     Producer {
         id: client_producer_id("me", "p1"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Client(ClientEpoch { attempt })),
-        }),
-        run_id: String::new(),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt }),
+        })),
     }
 }
 
@@ -774,10 +773,10 @@ async fn a_stale_producer_cant_close_the_stream() {
 fn workflow(run: &str) -> Producer {
     Producer {
         id: workflow_producer_id(run),
-        run_id: run.to_string(),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())),
-        }),
+        kind: Some(Kind::Workflow(WorkflowProducer {
+            run_id: run.to_string(),
+            epoch: Some(WorkflowTaskEpoch::default()),
+        })),
     }
 }
 
@@ -800,13 +799,14 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() {
 async fn a_producer_with_no_epoch_case_is_refused() {
     for target in stores_for(Needs::CALLS).await {
         target.create(Owner::Client).await;
-        for epoch in [None, Some(ProducerEpoch::default())] {
-            let producer = Producer { epoch, ..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);
-        }
+        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());
     }
 }
@@ -977,7 +977,7 @@ async fn a_producer_field_has_the_shared_layout() {
         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.epoch.unwrap_or_default());
+            let epoch = |producer: &Producer| epoch_key(producer.kind.as_ref());
             ProducerField {
                 fence: epoch(producer),
                 held: held.map(epoch),
diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs
index c0822051a..3e035a932 100644
--- a/crates/streams/src/testing/support.rs
+++ b/crates/streams/src/testing/support.rs
@@ -10,9 +10,9 @@ use crate::{
     epoch::withdraws,
     ids::local_activity_producer_id,
     proto::{
-        AppendRecord, LocalActivityEpoch, Producer, ProducerEpoch, StreamData, StreamRecord,
-        StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer_epoch::Kind,
-        stream_record::Item,
+        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},
 };
@@ -132,13 +132,13 @@ impl Run {
         };
         Producer {
             id: local_activity_producer_id(&self.id, call),
-            epoch: Some(ProducerEpoch {
-                kind: Some(Kind::LocalActivity(LocalActivityEpoch {
+            kind: Some(Kind::LocalActivity(LocalActivityProducer {
+                run_id: self.id.clone(),
+                epoch: Some(LocalActivityEpoch {
                     task: Some(task),
                     attempt,
-                })),
-            }),
-            run_id: self.id.clone(),
+                }),
+            })),
         }
     }
 }
@@ -193,11 +193,14 @@ pub(crate) fn digest(texts: &[&str]) -> [u8; crate::digest::DIGEST_LEN] {
 }
 
 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()),
-            withdraw_run: withdrawal.withdraw_run.to_string(),
-            ..Default::default()
+            withdrawal: run,
         }),
     }
     .encode_to_vec()
diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs
new file mode 100644
index 000000000..c509f688d
--- /dev/null
+++ b/crates/streams/src/testing/writer.rs
@@ -0,0 +1,758 @@
+//! The writers of Activities and clients, through the runtime's byte call, as lang sends them.
+//!
+//! Every Store runs the suite through its own runtime.
+
+use super::{
+    case,
+    support::{Needs, stores_for},
+};
+use crate::{
+    MemoryStreamRuntime, StreamError, StreamResult, StreamRuntime,
+    epoch::withdraws,
+    ids::{
+        activity_producer_id, client_producer_id, local_activity_producer_id,
+        named_client_producer_id,
+    },
+    proto::{
+        ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer,
+        CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CreateStreamRequest,
+        CreateStreamResponse, LocalActivityEpoch, LocalActivityProducer, OpenWriterRequest,
+        OpenWriterResponse, Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal,
+        StreamErrorDetails, StreamRecord, StreamRef, StreamWriter, WorkflowTaskEpoch,
+        WriteRecordsRequest, WriteRecordsResponse, producer::Kind, stream_record::Item,
+    },
+    store::{
+        Appended, Batch, Capabilities, Closed, Create, Created, Entry, MAX_BATCH_RECORDS, Store,
+        StreamState,
+    },
+};
+use prost::Message;
+use prost_types::Timestamp;
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use temporalio_common::protos::temporal::api::common::v1::Payload;
+use tonic::Code;
+
+static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+struct Case {
+    ext: Arc<StreamRuntime>,
+    namespace: String,
+    stream: String,
+}
+
+async fn cases() -> Vec<Case> {
+    let mut cases = vec![];
+    for ext in super::support::stream_runtimes() {
+        let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let case = Case {
+            ext,
+            namespace: format!("ns-{now}-{n}"),
+            stream: format!("s-{now}-{n}"),
+        };
+        let created: CreateStreamResponse = case
+            .call(
+                "CreateStream",
+                CreateStreamRequest {
+                    namespace: case.namespace.clone(),
+                    stream_ref: Some(StreamRef {
+                        id: case.stream.clone(),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert!(created.created);
+        cases.push(case);
+    }
+    cases
+}
+
+/// The writer of a write by `producer`, with no withdrawal facts.
+fn writer_for(producer: Producer) -> Option<StreamWriter> {
+    Some(StreamWriter {
+        producer: Some(producer),
+        ..Default::default()
+    })
+}
+
+fn client() -> Producer {
+    Producer {
+        id: client_producer_id("me", "p"),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt: 1 }),
+        })),
+    }
+}
+
+fn activity(attempt: i32) -> Producer {
+    Producer {
+        id: activity_producer_id("r1", "a", "s"),
+        kind: Some(Kind::Activity(ActivityProducer {
+            run_id: "r1".to_string(),
+            epoch: Some(ActivityEpoch {
+                attempt_scheduled_time: Some(Timestamp {
+                    seconds: 10 * attempt as i64,
+                    nanos: 0,
+                }),
+                attempt,
+            }),
+        })),
+    }
+}
+
+fn records(texts: &[&str]) -> Vec<AppendRecord> {
+    texts
+        .iter()
+        .map(|text| AppendRecord {
+            body: Some(Payload {
+                data: text.as_bytes().to_vec(),
+                ..Default::default()
+            }),
+        })
+        .collect()
+}
+
+fn reason(error: &StreamError) -> String {
+    let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap();
+    assert_eq!(details.reason(), error.reason.proto());
+    error.reason.token().to_string()
+}
+
+fn millis(ms: u64) -> Option<prost_types::Duration> {
+    Some(prost_types::Duration {
+        seconds: (ms / 1000) as i64,
+        nanos: ((ms % 1000) * 1_000_000) as i32,
+    })
+}
+
+impl Case {
+    async fn call<Req: Message, Resp: Message + Default>(
+        &self,
+        method: &str,
+        request: Req,
+    ) -> Result<Resp, StreamError> {
+        let bytes = self.ext.call(method, &request.encode_to_vec()).await?;
+        Ok(Resp::decode(bytes.as_slice()).unwrap())
+    }
+
+    async fn open(&self, producer: Producer, change: impl Fn(&mut OpenWriterRequest)) -> u64 {
+        let mut request = OpenWriterRequest {
+            namespace: self.namespace.clone(),
+            r#ref: Some(StreamRef {
+                id: self.stream.clone(),
+                ..Default::default()
+            }),
+            writer: writer_for(producer),
+            ..Default::default()
+        };
+        change(&mut request);
+        let opened: OpenWriterResponse = self.call("OpenWriter", request).await.unwrap();
+        opened.writer_id
+    }
+
+    /// An awaited publish. The Records are in the Store when it returns.
+    async fn write(&self, writer_id: u64, texts: &[&str]) -> Result<(i64, i64), StreamError> {
+        let written: WriteRecordsResponse = self
+            .call(
+                "WriteRecords",
+                WriteRecordsRequest {
+                    writer_id,
+                    records: records(texts),
+                },
+            )
+            .await?;
+        Ok((written.first_offset, written.last_offset))
+    }
+
+    async fn read(&self, wait_ms: u64) -> ReadRecordsResponse {
+        self.call(
+            "ReadRecords",
+            ReadRecordsRequest {
+                namespace: self.namespace.clone(),
+                r#ref: Some(StreamRef {
+                    id: self.stream.clone(),
+                    ..Default::default()
+                }),
+                wait: millis(wait_ms),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap()
+    }
+
+    async fn visible(&self) -> Vec<String> {
+        let mut kept: Vec<StreamRecord> = vec![];
+        for record in self.read(0).await.records {
+            match &record.item {
+                Some(Item::Superseded(s)) => kept.retain(|old| !withdraws(s, old)),
+                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()
+    }
+}
+
+async fn a_write_is_in_the_store_when_it_answers_its_offsets() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        let (first, last) = case.write(writer, &["a", "b"]).await.unwrap();
+        assert!(0 < first && first < last);
+        let read = case.read(0).await;
+        let offsets: Vec<i64> = read.records.iter().map(|r| r.offset).collect();
+        assert_eq!(offsets, [first, last]);
+        let (one, same) = case.write(writer, &["c"]).await.unwrap();
+        assert_eq!(one, same);
+        assert!(one > last);
+        assert_eq!(case.write(writer, &[]).await.unwrap(), (0, 0));
+        assert_eq!(case.visible().await, ["a", "b", "c"]);
+        let closed: CloseWriterResponse = case
+            .call("CloseWriter", CloseWriterRequest { writer_id: writer })
+            .await
+            .unwrap();
+        assert_eq!(closed.last_offset, one);
+    }
+}
+
+async fn writes_split_at_the_batch_size_and_keep_one_sequence() {
+    for case in cases().await {
+        let writer = case.open(client(), |r| r.max_batch_size = 2).await;
+        let (first, last) = case.write(writer, &["1", "2", "3"]).await.unwrap();
+        case.write(writer, &["4", "5"]).await.unwrap();
+        let read = case.read(0).await;
+        assert_eq!(case.visible().await, ["1", "2", "3", "4", "5"]);
+        assert_eq!(read.records[0].offset, first);
+        assert_eq!(read.records[2].offset, last);
+        // Opening the same Producer epoch again returns the same writer, so it keeps the sequence.
+        assert_eq!(case.open(client(), |_| {}).await, writer);
+    }
+}
+
+async fn a_batch_size_over_the_store_limit_writes_in_more_batches() {
+    for case in cases().await {
+        let writer = case.open(client(), |r| r.max_batch_size = 5000).await;
+        let texts: Vec<String> = (0..=MAX_BATCH_RECORDS).map(|i| i.to_string()).collect();
+        let texts: Vec<&str> = texts.iter().map(String::as_str).collect();
+        // The Store refuses one batch of this many, so the write goes in two.
+        let (first, last) = case.write(writer, &texts).await.unwrap();
+        assert!(first < last);
+    }
+}
+
+async fn a_failed_write_fails_every_later_write_and_the_close() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        let close = CloseStreamRequest {
+            namespace: case.namespace.clone(),
+            r#ref: Some(StreamRef {
+                id: case.stream.clone(),
+                ..Default::default()
+            }),
+            writer: writer_for(client()),
+            ..Default::default()
+        };
+        let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap();
+        let error = case.write(writer, &["late"]).await.unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::FailedPrecondition, "stream_closed")
+        );
+        assert_eq!(
+            reason(&case.write(writer, &["x"]).await.unwrap_err()),
+            "stream_closed"
+        );
+        let error = case
+            .call::<_, CloseWriterResponse>("CloseWriter", CloseWriterRequest { writer_id: writer })
+            .await
+            .unwrap_err();
+        assert_eq!(reason(&error), "stream_closed");
+    }
+}
+
+async fn an_activity_retry_withdraws_the_earlier_attempt_even_with_nothing_to_write() {
+    for case in cases().await {
+        let first = case.open(activity(1), |_| {}).await;
+        case.write(first, &["a1"]).await.unwrap();
+        case.open(activity(2), |_| {}).await;
+        assert!(case.visible().await.is_empty());
+    }
+}
+
+/// An Activity retry opens its writer on a closed or expired Stream, so a retry that only reads
+/// the Stream still runs. Its first write fails with the real reason.
+async fn an_activity_retry_opens_on_a_closed_or_expired_stream() {
+    for case in cases().await {
+        let first = case.open(activity(1), |_| {}).await;
+        case.write(first, &["a1"]).await.unwrap();
+        let close = CloseStreamRequest {
+            namespace: case.namespace.clone(),
+            r#ref: Some(StreamRef {
+                id: case.stream.clone(),
+                ..Default::default()
+            }),
+            writer: writer_for(client()),
+            ..Default::default()
+        };
+        let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap();
+        let retry = case.open(activity(2), |_| {}).await;
+        let error = case.write(retry, &["a2"]).await.unwrap_err();
+        assert_eq!(reason(&error), "stream_closed");
+
+        let short = format!("{}-short", case.stream);
+        let _: CreateStreamResponse = case
+            .call(
+                "CreateStream",
+                CreateStreamRequest {
+                    namespace: case.namespace.clone(),
+                    stream_ref: Some(StreamRef {
+                        id: short.clone(),
+                        ..Default::default()
+                    }),
+                    retention: millis(300),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        let on_short = |request: &mut OpenWriterRequest| {
+            request.r#ref = Some(StreamRef {
+                id: short.clone(),
+                ..Default::default()
+            });
+        };
+        let first = case.open(activity(1), on_short).await;
+        case.write(first, &["a1"]).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(500)).await;
+        let retry = case.open(activity(2), on_short).await;
+        let error = case.write(retry, &["a2"]).await.unwrap_err();
+        assert_eq!(reason(&error), "stream_expired");
+    }
+}
+
+async fn a_closed_writer_is_gone() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        case.write(writer, &["last"]).await.unwrap();
+        let closed: CloseWriterResponse = case
+            .call("CloseWriter", CloseWriterRequest { writer_id: writer })
+            .await
+            .unwrap();
+        assert!(closed.last_offset > 0);
+        let error = case.write(writer, &["x"]).await.unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::NotFound, "stream_unknown_writer")
+        );
+    }
+}
+
+async fn a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        let big = "x".repeat(1_400_000);
+        case.write(writer, &[&big, &big, &big, "end"])
+            .await
+            .unwrap();
+        assert_eq!(case.visible().await.len(), 4);
+        let huge = "y".repeat(1_700_000);
+        let error = case.write(writer, &[&huge]).await.unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::InvalidArgument, "stream_too_large")
+        );
+        // A refusal before the Store leaves the writer usable.
+        case.write(writer, &["ok"]).await.unwrap();
+    }
+}
+
+async fn a_reopen_of_a_closed_writer_goes_on_with_its_sequence() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        case.write(writer, &["1"]).await.unwrap();
+        let _: CloseWriterResponse = case
+            .call("CloseWriter", CloseWriterRequest { writer_id: writer })
+            .await
+            .unwrap();
+        let again = case.open(client(), |_| {}).await;
+        assert_ne!(again, writer);
+        case.write(again, &["2"]).await.unwrap();
+        assert_eq!(case.visible().await, ["1", "2"]);
+    }
+}
+
+async fn the_opens_of_one_producer_share_a_writer_until_the_last_one_lets_go() {
+    for case in cases().await {
+        let a = case.open(client(), |_| {}).await;
+        let b = case.open(client(), |_| {}).await;
+        assert_eq!(a, b);
+        case.write(a, &["1"]).await.unwrap();
+        let _: CloseWriterResponse = case
+            .call("CloseWriter", CloseWriterRequest { writer_id: a })
+            .await
+            .unwrap();
+        // The other reference still writes, on the same sequence.
+        case.write(b, &["2"]).await.unwrap();
+        case.ext.free_writer(b).unwrap();
+        let error = case.write(b, &["3"]).await.unwrap_err();
+        assert_eq!(reason(&error), "stream_unknown_writer");
+        assert_eq!(case.visible().await, ["1", "2"]);
+    }
+}
+
+async fn a_named_client_with_a_higher_epoch_fences_its_earlier_self() {
+    for case in cases().await {
+        let named = |attempt| Producer {
+            id: named_client_producer_id("ingest-1"),
+            kind: Some(Kind::Client(ClientProducer {
+                epoch: Some(ClientEpoch { attempt }),
+            })),
+        };
+        let old = case.open(named(1), |_| {}).await;
+        case.write(old, &["old"]).await.unwrap();
+        let new = case.open(named(2), |_| {}).await;
+        case.write(new, &["new"]).await.unwrap();
+        let error = case.write(old, &["late"]).await.unwrap_err();
+        assert_eq!(reason(&error), "stream_fenced");
+        // The earlier epoch's Records are real messages, so the new epoch withdraws none.
+        assert_eq!(case.visible().await, ["old", "new"]);
+    }
+}
+
+async fn closing_every_writer_drops_them_all() {
+    for case in cases().await {
+        let a = case.open(client(), |_| {}).await;
+        let b = case.open(activity(1), |_| {}).await;
+        case.write(a, &["a"]).await.unwrap();
+        case.write(b, &["b"]).await.unwrap();
+        let closed: crate::proto::CloseWritersResponse = case
+            .call("CloseWriters", crate::proto::CloseWritersRequest {})
+            .await
+            .unwrap();
+        assert_eq!(closed.failed, 0);
+        assert!(case.write(a, &["x"]).await.is_err());
+        assert!(case.write(b, &["x"]).await.is_err());
+    }
+}
+
+/// A Store over another one, whose appends fail. It can write an append and lose its answer, as on
+/// a dropped connection, and it refuses every append while it is down. It can also refuse the next
+/// append as corrupt Store data.
+struct Flaky {
+    inner: Arc<dyn Store>,
+    lose_next: AtomicBool,
+    down: AtomicBool,
+    corrupt_next: AtomicBool,
+}
+
+impl Flaky {
+    /// A Store that is up, over `inner`.
+    fn new(inner: Arc<dyn Store>) -> Self {
+        Self {
+            inner,
+            lose_next: AtomicBool::new(false),
+            down: AtomicBool::new(false),
+            corrupt_next: AtomicBool::new(false),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl Store for Flaky {
+    fn capabilities(&self) -> Capabilities {
+        self.inner.capabilities()
+    }
+
+    async fn create(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        create: &Create<'_>,
+    ) -> StreamResult<Created> {
+        self.inner.create(namespace, stream_ref, create).await
+    }
+
+    async fn append(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        batch: &Batch<'_>,
+    ) -> StreamResult<Appended> {
+        if self.down.load(Ordering::SeqCst) {
+            return Err(StreamError::storage_unavailable("the Store is down"));
+        }
+        if self.corrupt_next.swap(false, Ordering::SeqCst) {
+            return Err(StreamError::storage_corrupt("a stored field is bad"));
+        }
+        let appended = self.inner.append(namespace, stream_ref, batch).await?;
+        if self.lose_next.swap(false, Ordering::SeqCst) {
+            return Err(StreamError::storage_unavailable("the answer was lost"));
+        }
+        Ok(appended)
+    }
+
+    async fn read(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        from: i64,
+        count: usize,
+    ) -> StreamResult<(StreamState, Vec<Entry>)> {
+        self.inner.read(namespace, stream_id, from, count).await
+    }
+
+    async fn wait_after(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        after: i64,
+        deadline: tokio::time::Instant,
+    ) -> StreamResult<bool> {
+        self.inner
+            .wait_after(namespace, stream_id, after, deadline)
+            .await
+    }
+
+    async fn close(
+        &self,
+        namespace: &str,
+        stream_ref: &StreamRef,
+        producer: &Producer,
+    ) -> StreamResult<Closed> {
+        self.inner.close(namespace, stream_ref, producer).await
+    }
+
+    async fn read_back(
+        &self,
+        namespace: &str,
+        stream_id: &str,
+        before: i64,
+        count: usize,
+    ) -> StreamResult<Vec<Entry>> {
+        self.inner
+            .read_back(namespace, stream_id, before, count)
+            .await
+    }
+}
+
+/// A case on each backend under test, on a [Flaky] Store over the backend's Store, with its Stream
+/// created.
+async fn flaky_cases(lose_next: bool) -> Vec<(Arc<Flaky>, Case)> {
+    let mut cases = vec![];
+    for target in stores_for(Needs::CALLS).await {
+        let store = Arc::new(Flaky {
+            lose_next: AtomicBool::new(lose_next),
+            ..Flaky::new(target.store)
+        });
+        let case = Case {
+            ext: MemoryStreamRuntime::with_store(store.clone(), None),
+            namespace: target.namespace,
+            stream: target.stream,
+        };
+        let _: CreateStreamResponse = case
+            .call(
+                "CreateStream",
+                CreateStreamRequest {
+                    namespace: case.namespace.clone(),
+                    stream_ref: Some(StreamRef {
+                        id: case.stream.clone(),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        cases.push((store, case));
+    }
+    cases
+}
+
+async fn a_write_whose_answer_is_lost_goes_in_once() {
+    for (_, case) in flaky_cases(true).await {
+        let writer = case.open(client(), |_| {}).await;
+        let (first, last) = case.write(writer, &["a", "b"]).await.unwrap();
+        let read = case.read(0).await;
+        assert_eq!(case.visible().await, ["a", "b"]);
+        assert_eq!(
+            (read.records[0].offset, read.records[1].offset),
+            (first, last)
+        );
+        case.write(writer, &["c"]).await.unwrap();
+        assert_eq!(case.visible().await, ["a", "b", "c"]);
+    }
+}
+
+/// A write that fails in a Store outage fails alone, and the next write succeeds once the Store is
+/// back.
+async fn a_writer_writes_again_after_a_store_outage() {
+    for (store, case) in flaky_cases(false).await {
+        let writer = case
+            .open(client(), |r| r.max_retry_duration = millis(100))
+            .await;
+        case.write(writer, &["a"]).await.unwrap();
+        store.down.store(true, Ordering::SeqCst);
+        let error = case.write(writer, &["lost"]).await.unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::Unavailable, "stream_storage_unavailable")
+        );
+        store.down.store(false, Ordering::SeqCst);
+        case.write(writer, &["b"]).await.unwrap();
+        assert_eq!(case.visible().await, ["a", "b"]);
+        let closed: CloseWriterResponse = case
+            .call("CloseWriter", CloseWriterRequest { writer_id: writer })
+            .await
+            .unwrap();
+        assert!(closed.last_offset > 0);
+    }
+}
+
+/// A write that meets corrupt Store data fails at once, since a retry reads the same data. The
+/// writer then fails every later write.
+async fn a_write_on_corrupt_store_data_fails_with_no_retry() {
+    for (store, case) in flaky_cases(false).await {
+        let writer = case.open(client(), |_| {}).await;
+        store.corrupt_next.store(true, Ordering::SeqCst);
+        let error = case.write(writer, &["a"]).await.unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::DataLoss, "stream_storage_corrupt")
+        );
+        assert!(case.visible().await.is_empty());
+        let again = case.write(writer, &["b"]).await.unwrap_err();
+        assert_eq!(reason(&again), "stream_storage_corrupt");
+    }
+}
+
+/// An open of a writer with a run to withdraw fails with `stream_invalid` and writes nothing, since
+/// only a retry run's first append withdraws a run and a writer sends many batches.
+async fn a_writer_with_a_run_to_withdraw_is_refused() {
+    for case in cases().await {
+        let request = OpenWriterRequest {
+            namespace: case.namespace.clone(),
+            r#ref: Some(StreamRef {
+                id: case.stream.clone(),
+                ..Default::default()
+            }),
+            writer: Some(StreamWriter {
+                producer: Some(client()),
+                withdrawal: Some(RunWithdrawal {
+                    withdraw_run: "r1".to_string(),
+                    ..Default::default()
+                }),
+            }),
+            ..Default::default()
+        };
+        let error = case
+            .call::<_, OpenWriterResponse>("OpenWriter", request)
+            .await
+            .unwrap_err();
+        assert_eq!(
+            (error.code(), reason(&error).as_str()),
+            (Code::InvalidArgument, "stream_invalid")
+        );
+        assert!(case.visible().await.is_empty());
+    }
+}
+
+async fn a_write_that_waits_behind_a_close_is_refused() {
+    for case in cases().await {
+        let writer = case.open(client(), |_| {}).await;
+        case.write(writer, &["a"]).await.unwrap();
+        let close = case.call::<_, CloseWriterResponse>(
+            "CloseWriter",
+            CloseWriterRequest { writer_id: writer },
+        );
+        let write = async {
+            tokio::task::yield_now().await;
+            case.write(writer, &["late"]).await
+        };
+        let (closed, written) = tokio::join!(close, write);
+        closed.unwrap();
+        assert_eq!(reason(&written.unwrap_err()), "stream_unknown_writer");
+        assert_eq!(case.visible().await, ["a"]);
+    }
+}
+
+async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() {
+    for case in cases().await {
+        let local_activity = |call: &str, started: i64| Producer {
+            id: local_activity_producer_id("r1", call),
+            kind: Some(Kind::LocalActivity(LocalActivityProducer {
+                run_id: "r1".to_string(),
+                epoch: Some(LocalActivityEpoch {
+                    task: Some(WorkflowTaskEpoch {
+                        started_event_id: started,
+                        attempt: 1,
+                        started_time: Some(Timestamp {
+                            seconds: started,
+                            nanos: 0,
+                        }),
+                    }),
+                    attempt: 1,
+                }),
+            })),
+        };
+        let facts = |completed: i64, failed: Vec<i64>| {
+            move |request: &mut OpenWriterRequest| {
+                let writer = request.writer.get_or_insert_with(Default::default);
+                let withdrawal = writer.withdrawal.get_or_insert_with(Default::default);
+                withdrawal.previous_completed_task = Some(WorkflowTaskEpoch {
+                    started_event_id: completed,
+                    ..Default::default()
+                });
+                withdrawal.failed_tasks = failed.clone();
+            }
+        };
+        // Task [3] fails after its Local Activity writes, and task [4] completes with no write.
+        let orphan = case.open(local_activity("1", 3), facts(2, vec![])).await;
+        case.write(orphan, &["orphan"]).await.unwrap();
+        let next = case.open(local_activity("2", 5), facts(4, vec![3])).await;
+        case.write(next, &["next"]).await.unwrap();
+        assert_eq!(case.visible().await, ["next"]);
+        // Task [5] completed while call 2 ran, and task [6], where the call resolved, failed.
+        // Call 2 runs again in task [8] and writes nothing. Its open withdraws its earlier task.
+        case.open(local_activity("2", 8), facts(5, vec![6])).await;
+        assert!(case.visible().await.is_empty());
+    }
+}
+
+/// The cases of the suite, in file order.
+pub const CASES: &[super::Case] = &[
+    case!(a_write_is_in_the_store_when_it_answers_its_offsets),
+    case!(writes_split_at_the_batch_size_and_keep_one_sequence),
+    case!(a_batch_size_over_the_store_limit_writes_in_more_batches),
+    case!(a_failed_write_fails_every_later_write_and_the_close),
+    case!(an_activity_retry_withdraws_the_earlier_attempt_even_with_nothing_to_write),
+    case!(an_activity_retry_opens_on_a_closed_or_expired_stream),
+    case!(a_closed_writer_is_gone),
+    case!(a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one),
+    case!(a_reopen_of_a_closed_writer_goes_on_with_its_sequence),
+    case!(the_opens_of_one_producer_share_a_writer_until_the_last_one_lets_go),
+    case!(a_named_client_with_a_higher_epoch_fences_its_earlier_self),
+    case!(closing_every_writer_drops_them_all),
+    case!(a_write_whose_answer_is_lost_goes_in_once),
+    case!(a_writer_writes_again_after_a_store_outage),
+    case!(a_write_on_corrupt_store_data_fails_with_no_retry),
+    case!(a_writer_with_a_run_to_withdraw_is_refused),
+    case!(a_write_that_waits_behind_a_close_is_refused),
+    case!(a_local_activity_writer_withdraws_the_records_of_a_failed_task),
+];
diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs
new file mode 100644
index 000000000..88e31342f
--- /dev/null
+++ b/crates/streams/src/writer.rs
@@ -0,0 +1,401 @@
+//! Writers for Activities and clients. `write` appends its Records at once and returns their
+//! offsets, as an awaited publish needs. A writer keeps one Producer epoch and one `sequence`
+//! order, so that the Store writes a retried batch only once. Lang can buffer Records on its own
+//! and write each batch with one call.
+//!
+//! The opens of one Producer epoch on one Stream share one writer. Each open adds a reference, and
+//! each close or free releases one. The last release drops the writer.
+
+use crate::{
+    Reason, StreamError, StreamResult,
+    epoch::{ProducerKind, writer_withdraw_run},
+    proto::{
+        AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamRef, StreamWriter,
+        producer::Kind,
+    },
+    service::StreamService,
+    store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES},
+};
+use parking_lot::Mutex;
+use prost::Message;
+use std::{
+    collections::HashMap,
+    sync::{
+        Arc,
+        atomic::{AtomicI64, AtomicU64, Ordering},
+    },
+    time::Duration,
+};
+use tokio::time::Instant;
+
+const DEFAULT_RETRY: Duration = Duration::from_secs(10);
+const FIRST_BACKOFF: Duration = Duration::from_millis(50);
+
+/// The last `sequence` of any batch in this process. Every writer numbers its batches from it, so
+/// that a writer that opens again after its close continues above its earlier batches.
+static LAST_SEQUENCE: AtomicI64 = AtomicI64::new(0);
+
+fn next_sequence() -> i64 {
+    LAST_SEQUENCE.fetch_add(1, Ordering::Relaxed) + 1
+}
+
+#[derive(Default)]
+struct State {
+    last_offset: i64,
+    /// A write that the Store refused, such as on a closed Stream or a newer epoch. Every later
+    /// call fails with it. A Store failure fails only its own write.
+    failed: Option<StreamError>,
+    closed: bool,
+}
+
+struct Writer {
+    namespace: String,
+    stream_ref: StreamRef,
+    producer: Producer,
+    /// The open's writer, with its Producer and task facts, which each batch sends.
+    stream_writer: StreamWriter,
+    max_batch: usize,
+    max_retry: Duration,
+    service: Arc<StreamService>,
+    state: Mutex<State>,
+    /// Allows one write at a time, so that the batches of one epoch reach the Store in order.
+    writing: tokio::sync::Mutex<()>,
+}
+
+fn duration(value: Option<prost_types::Duration>) -> Option<Duration> {
+    let value = value?;
+    let duration = Duration::new(value.seconds.max(0) as u64, value.nanos.max(0) as u32);
+    (!duration.is_zero()).then_some(duration)
+}
+
+impl Writer {
+    /// The number of `records` in the next batch. The Store refuses a batch over its limit, so a
+    /// batch ends at the count limit or the byte limit, whichever comes first.
+    fn batch_len<'a>(&self, records: impl Iterator<Item = &'a AppendRecord>) -> usize {
+        let mut count = 0;
+        let mut bytes = 0;
+        for record in records.take(self.max_batch) {
+            let size = stored_size(&self.producer, record);
+            if count > 0 && bytes + size > MAX_BATCH_BYTES {
+                break;
+            }
+            bytes += size;
+            count += 1;
+        }
+        count
+    }
+
+    /// Appends `records` now and returns the offsets of the first and the last one. Records past
+    /// the Store's batch limit go in more batches, in order.
+    async fn write(&self, mut records: Vec<AppendRecord>) -> StreamResult<(i64, i64)> {
+        let _writing = self.writing.lock().await;
+        // A close that started while this write waited has ended the writer.
+        if self.state.lock().closed {
+            return Err(StreamError::new(
+                Reason::UnknownWriter,
+                "The writer closed while the write waited",
+            ));
+        }
+        let (mut first, mut last) = (0, 0);
+        while !records.is_empty() {
+            let count = self.batch_len(records.iter());
+            let rest = records.split_off(count);
+            if let Some(error) = &self.state.lock().failed {
+                return Err(error.clone());
+            }
+            let sequence = next_sequence();
+            let (batch_first, batch_last) = match self.send(records, sequence).await {
+                Ok(offsets) => offsets,
+                // The next write takes a higher `sequence`, so the Store refuses a late copy of
+                // this batch after it.
+                Err(error) if error.reason == Reason::StorageUnavailable => return Err(error),
+                Err(error) => {
+                    self.state.lock().failed = Some(error.clone());
+                    return Err(error);
+                }
+            };
+            if first == 0 {
+                first = batch_first;
+            }
+            last = batch_last;
+            records = rest;
+        }
+        let mut state = self.state.lock();
+        state.last_offset = state.last_offset.max(last);
+        Ok((first, last))
+    }
+
+    /// Appends one batch and returns its first and last offset. After a Store failure it sends the
+    /// batch again with the same `sequence`, so the dedupe key stops a second write.
+    async fn send(&self, records: Vec<AppendRecord>, sequence: i64) -> StreamResult<(i64, i64)> {
+        let request = AppendRecordsRequest {
+            namespace: self.namespace.clone(),
+            r#ref: Some(self.stream_ref.clone()),
+            writer: Some(self.stream_writer.clone()),
+            sequence,
+            records,
+            ..Default::default()
+        };
+        let until = Instant::now() + self.max_retry;
+        let mut backoff = FIRST_BACKOFF;
+        loop {
+            match self.service.append_records(request.clone()).await {
+                Ok(response) => return Ok((response.first_offset, response.last_offset)),
+                Err(error)
+                    if error.reason == Reason::StorageUnavailable
+                        && Instant::now() + backoff < until =>
+                {
+                    tokio::time::sleep(backoff).await;
+                    backoff = (backoff * 2).min(Duration::from_secs(1));
+                }
+                Err(error) => return Err(error),
+            }
+        }
+    }
+}
+
+/// The size of a Record once stored, with a margin for the envelope's own fields.
+fn stored_size(producer: &Producer, record: &AppendRecord) -> usize {
+    record.encoded_len() + producer.encoded_len() + 32
+}
+
+/// Refuses a Record over the Store's Record limit, before it reaches the Store.
+fn check_sizes(producer: &Producer, records: &[AppendRecord]) -> StreamResult<()> {
+    if let Some(big) = records
+        .iter()
+        .map(|record| stored_size(producer, record))
+        .find(|size| *size > MAX_RECORD_BYTES)
+    {
+        return Err(StreamError::new(
+            Reason::TooLarge,
+            format!("A Record has about {big} bytes, more than {MAX_RECORD_BYTES}"),
+        ));
+    }
+    Ok(())
+}
+
+/// The namespace, Stream id, Producer id and epoch key of a writer.
+type WriterKey = (String, String, String, [u8; crate::epoch::EPOCH_KEY_LEN]);
+
+/// One open writer and the references to it.
+struct Open {
+    writer: Arc<Writer>,
+    key: WriterKey,
+    refs: usize,
+}
+
+#[derive(Default)]
+struct Table {
+    open: HashMap<u64, Open>,
+    /// The writer of each key, so a second open shares its sequence instead of reusing it.
+    by_producer: HashMap<WriterKey, u64>,
+}
+
+impl Table {
+    /// Releases one reference to a writer, or every reference with `all`. Returns the writer when
+    /// it has no reference left, so that the caller ends it.
+    fn release(&mut self, id: u64, all: bool) -> StreamResult<Option<Arc<Writer>>> {
+        let open = self
+            .open
+            .get_mut(&id)
+            .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}")))?;
+        open.refs -= 1;
+        if open.refs > 0 && !all {
+            return Ok(None);
+        }
+        let open = self.open.remove(&id).expect("found above");
+        self.by_producer.remove(&open.key);
+        Ok(Some(open.writer))
+    }
+}
+
+/// The open writers of one runtime.
+#[derive(Default)]
+pub(crate) struct Writers {
+    next: AtomicU64,
+    table: Mutex<Table>,
+}
+
+impl Writers {
+    /// Opens a writer, or adds a reference to the one that this Producer epoch has on the Stream. A
+    /// writer of a Local Activity, or of an Activity attempt above 1, first sends an empty batch.
+    /// So the new epoch withdraws the earlier one's Records even when it publishes nothing. On a
+    /// closed or expired Stream the open still succeeds, and the first write fails with the reason.
+    pub(crate) async fn open(
+        &self,
+        service: &Arc<StreamService>,
+        request: OpenWriterRequest,
+    ) -> StreamResult<u64> {
+        let stream = request
+            .r#ref
+            .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?;
+        crate::ids::check_stream_id(&stream.id)?;
+        let stream_writer = request.writer.unwrap_or_default();
+        // Only a retry run's first append withdraws a run, and a writer sends many batches.
+        if !writer_withdraw_run(&stream_writer).is_empty() {
+            return Err(StreamError::invalid("A writer doesn't withdraw a run"));
+        }
+        let producer = stream_writer
+            .producer
+            .clone()
+            .ok_or_else(|| StreamError::invalid("The writer has no Producer"))?;
+        crate::ids::check_producer(&producer)?;
+        if ProducerKind::try_from(&producer)? == ProducerKind::Workflow {
+            return Err(StreamError::invalid("Workflow code doesn't open a writer"));
+        }
+        let key = (
+            request.namespace.clone(),
+            stream.id.clone(),
+            producer.id.clone(),
+            crate::epoch::epoch_key(producer.kind.as_ref()),
+        );
+        if let Some(id) = self.share(&key) {
+            return Ok(id);
+        }
+        // A Local Activity that runs again in a later task has attempt 1 again, so its writer
+        // always sends the empty batch.
+        let retry = match producer.kind.as_ref() {
+            Some(Kind::LocalActivity(_)) => true,
+            Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => {
+                activity.epoch.map_or(0, |e| e.attempt) > 1
+            }
+            _ => false,
+        };
+        let writer = Arc::new(Writer {
+            namespace: request.namespace,
+            stream_ref: stream,
+            producer,
+            stream_writer,
+            // The Store refuses a batch over its Record limit, so no batch can be larger.
+            max_batch: match request.max_batch_size {
+                n if n > 0 => (n as usize).min(MAX_BATCH_RECORDS),
+                _ => MAX_BATCH_RECORDS,
+            },
+            max_retry: duration(request.max_retry_duration).unwrap_or(DEFAULT_RETRY),
+            service: service.clone(),
+            state: Mutex::default(),
+            writing: tokio::sync::Mutex::new(()),
+        });
+        if retry {
+            match writer.send(vec![], next_sequence()).await {
+                // A Stream with no key has nothing to withdraw yet. A closed or expired one accepts
+                // no more Records, and a retry that only reads it must still run.
+                Ok(_) => {}
+                Err(error)
+                    if matches!(
+                        error.reason,
+                        Reason::NotFound | Reason::Closed | Reason::Expired
+                    ) => {}
+                Err(error) => return Err(error),
+            }
+        }
+        let mut table = self.table.lock();
+        if let Some(id) = table.by_producer.get(&key).copied() {
+            table.open.get_mut(&id).expect("indexed").refs += 1;
+            return Ok(id);
+        }
+        let id = self.next.fetch_add(1, Ordering::Relaxed) + 1;
+        table.by_producer.insert(key.clone(), id);
+        table.open.insert(
+            id,
+            Open {
+                writer,
+                key,
+                refs: 1,
+            },
+        );
+        Ok(id)
+    }
+
+    /// Adds a reference to the open writer of `key`, if there is one.
+    fn share(&self, key: &WriterKey) -> Option<u64> {
+        let mut table = self.table.lock();
+        let id = table.by_producer.get(key).copied()?;
+        table.open.get_mut(&id).expect("indexed").refs += 1;
+        Some(id)
+    }
+
+    fn get(&self, id: u64) -> StreamResult<Arc<Writer>> {
+        self.table
+            .lock()
+            .open
+            .get(&id)
+            .map(|open| open.writer.clone())
+            .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}")))
+    }
+
+    /// Appends Records at once and returns the first and the last offset. A retry after a Store
+    /// failure resends the same batch, so that the Store writes it once.
+    pub(crate) async fn write(
+        &self,
+        id: u64,
+        records: Vec<AppendRecord>,
+    ) -> StreamResult<(i64, i64)> {
+        let writer = self.get(id)?;
+        {
+            let state = writer.state.lock();
+            if let Some(error) = &state.failed {
+                return Err(error.clone());
+            }
+            if state.closed {
+                return Err(StreamError::new(
+                    Reason::UnknownWriter,
+                    format!("Writer {id} is closed"),
+                ));
+            }
+        }
+        check_sizes(&writer.producer, &records)?;
+        writer.write(records).await
+    }
+
+    /// Closes every open writer, whatever its references. Returns how many had failed.
+    pub(crate) async fn close_all(&self) -> usize {
+        let ids: Vec<u64> = self.table.lock().open.keys().copied().collect();
+        let mut failed = 0;
+        for id in ids {
+            let released = self.table.lock().release(id, true);
+            let Ok(Some(writer)) = released else {
+                continue;
+            };
+            if end(&writer).await.is_err() {
+                failed += 1;
+            }
+        }
+        failed
+    }
+
+    /// Releases one reference. The last one waits for a write in flight, then drops the writer.
+    /// Returns its last offset, or the error it failed with.
+    pub(crate) async fn close(&self, id: u64) -> StreamResult<i64> {
+        let writer = self.get(id)?;
+        if self.table.lock().release(id, false)?.is_none() {
+            let state = writer.state.lock();
+            return match &state.failed {
+                Some(error) => Err(error.clone()),
+                None => Ok(state.last_offset),
+            };
+        }
+        end(&writer).await
+    }
+
+    /// Releases one reference with no wait, for a reference that lang drops. A write in flight
+    /// still completes, and a write that waits behind it is refused once no reference is left.
+    pub(crate) fn free(&self, id: u64) -> StreamResult<()> {
+        if let Some(writer) = self.table.lock().release(id, false)? {
+            writer.state.lock().closed = true;
+        }
+        Ok(())
+    }
+}
+
+/// Ends a writer that no reference uses. It waits for a write in flight, then returns its last
+/// offset or the error it failed with.
+async fn end(writer: &Writer) -> StreamResult<i64> {
+    writer.state.lock().closed = true;
+    let _writing = writer.writing.lock().await;
+    let state = writer.state.lock();
+    match &state.failed {
+        Some(error) => Err(error.clone()),
+        None => Ok(state.last_offset),
+    }
+}
diff --git a/crates/streams/tests/capabilities.rs b/crates/streams/tests/capabilities.rs
index 2b312f1da..34212ccc9 100644
--- a/crates/streams/tests/capabilities.rs
+++ b/crates/streams/tests/capabilities.rs
@@ -12,10 +12,10 @@ use temporalio_streams::{
     ids::client_producer_id,
     memory::MemoryStore,
     proto::{
-        AppendRecord, AppendRecordsRequest, ClientEpoch, CloseStreamRequest, CreateStreamRequest,
-        DescribeStreamRequest, DescribeStreamResponse, Producer, ProducerEpoch, StreamErrorDetails,
-        StreamErrorReason, StreamRef, StreamStatus, StreamWriter, WorkflowTaskEpoch,
-        producer_epoch::Kind,
+        AppendRecord, AppendRecordsRequest, ClientEpoch, ClientProducer, CloseStreamRequest,
+        CreateStreamRequest, DescribeStreamRequest, DescribeStreamResponse, Producer,
+        StreamErrorDetails, StreamErrorReason, StreamRef, StreamStatus, StreamWriter,
+        WorkflowProducer, WorkflowTaskEpoch, producer::Kind,
     },
     store::{Appended, Batch, Capabilities, Closed, Create, Created, Entry, Store, StreamState},
 };
@@ -120,10 +120,9 @@ fn case(without: Capabilities) -> Case {
 fn client() -> Producer {
     Producer {
         id: client_producer_id("me", "p"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Client(ClientEpoch { attempt: 1 })),
-        }),
-        run_id: String::new(),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt: 1 }),
+        })),
     }
 }
 
@@ -186,10 +185,10 @@ impl Case {
             withdraw_run: withdraw_run.to_string(),
             owner: Some(Producer {
                 id: temporalio_streams::ids::workflow_producer_id(&self.run_id),
-                epoch: Some(ProducerEpoch {
-                    kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())),
-                }),
-                run_id: self.run_id.clone(),
+                kind: Some(Kind::Workflow(WorkflowProducer {
+                    run_id: self.run_id.clone(),
+                    epoch: Some(WorkflowTaskEpoch::default()),
+                })),
             }),
             stream_ref: Some(StreamRef {
                 id: self.stream.clone(),
diff --git a/crates/streams/tests/scans.rs b/crates/streams/tests/scans.rs
index bcc5ea044..e2e5f54c4 100644
--- a/crates/streams/tests/scans.rs
+++ b/crates/streams/tests/scans.rs
@@ -14,10 +14,10 @@ use temporalio_streams::{
     ids::client_producer_id,
     memory::MemoryStore,
     proto::{
-        AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch,
-        CreateStreamRequest, CreateStreamResponse, Producer, ProducerEpoch, ReadRecordsRequest,
+        AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, ClientProducer,
+        CreateStreamRequest, CreateStreamResponse, Producer, ReadRecordsRequest,
         ReadRecordsResponse, StreamErrorDetails, StreamErrorReason, StreamRef, StreamWriter,
-        producer_epoch::Kind,
+        producer::Kind,
     },
     store::{Appended, Batch, Capabilities, Create, Created, Entry, Store, StreamState},
 };
@@ -129,10 +129,9 @@ async fn create(ext: &StreamRuntime) {
 fn client() -> Producer {
     Producer {
         id: client_producer_id("me", "p"),
-        epoch: Some(ProducerEpoch {
-            kind: Some(Kind::Client(ClientEpoch { attempt: 1 })),
-        }),
-        run_id: String::new(),
+        kind: Some(Kind::Client(ClientProducer {
+            epoch: Some(ClientEpoch { attempt: 1 }),
+        })),
     }
 }
 
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()
             })),