diff --git a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto
index a9b835480..90d75ef86 100644
--- a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto
+++ b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto
@@ -1,4 +1,4 @@
-// Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+// Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 // For the draft, every message and enum is in one package. A real change would split them
 // into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 syntax = "proto3";
diff --git a/scripts/_proto/temporal/sdk/streams/v1/extension.proto b/scripts/_proto/temporal/sdk/streams/v1/extension.proto
index 5791ae770..5945df743 100644
--- a/scripts/_proto/temporal/sdk/streams/v1/extension.proto
+++ b/scripts/_proto/temporal/sdk/streams/v1/extension.proto
@@ -87,42 +87,31 @@ message FlushRunBufferResponse {
     int32 records = 1;
 }
 
-// Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer.
+// Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps
+// its epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty
+// batch, so the new epoch withdraws the earlier attempt's Records.
 message OpenWriterRequest {
     string namespace = 1;
     temporal.api.streamservice.v1.StreamRef stream = 2;
     temporal.api.streamservice.v1.Producer producer = 3;
-    // How often the writer flushes. Empty is 100 ms.
-    google.protobuf.Duration batch_interval = 4;
-    // A flush sends at most this many Records in one batch. Zero is 1000.
+    // The interval flush and its buffer limit are gone. Lang buffers on its own.
+    reserved 4, 7;
+    reserved "batch_interval", "max_buffered_records";
+    // A write sends at most this many Records in one batch. Zero is 1000.
     int32 max_batch_size = 5;
-    // How long a flush retries a Store failure. Empty is 10 s.
+    // How long a write retries a Store failure. Empty is 10 s.
     google.protobuf.Duration max_retry_duration = 6;
-    // A publish past this many buffered Records fails with `stream_buffer_full`. Zero is 10000.
-    int32 max_buffered_records = 7;
 }
 
 message OpenWriterResponse {
     uint64 writer_id = 1;
 }
 
-// Method `Publish`, synchronous. Adds encoded Records to a writer's buffer.
-message PublishRequest {
-    uint64 writer_id = 1;
-    repeated temporal.api.streamservice.v1.AppendRecord records = 2;
-    // Start a flush at once.
-    bool force_flush = 3;
-}
-
-message PublishResponse {
-}
-
-// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer, after the
-// Records that `Publish` buffered, and answers when they are in the Store. It keeps the writer's
-// Producer epoch and `sequence` order, and a Store failure resends the same batch, so it goes in
-// once. Records past the Store's batch limit go as more batches, in order. An awaited publish of
-// an Activity or a client uses this call, so it needs no interval flush. A failure after the
-// Store retries fails the writer, as a failed flush does.
+// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and answers
+// when they are in the Store. It keeps the writer's Producer epoch and `sequence` order, and a
+// Store failure resends the same batch, so it goes in once. Records past the Store's batch limit
+// go as more batches, in order. An awaited publish of an Activity or a client uses this call. A
+// failure after the Store retries fails the writer, since the batch may be in the Store.
 message WriteRecordsRequest {
     uint64 writer_id = 1;
     repeated temporal.api.streamservice.v1.AppendRecord records = 2;
@@ -134,33 +123,24 @@ message WriteRecordsResponse {
     int64 last_offset = 2;
 }
 
-// Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store.
-message FlushWriterRequest {
-    uint64 writer_id = 1;
-}
-
-message FlushWriterResponse {
-    // The offset of the newest Record the writer wrote. Zero if it wrote none.
-    int64 last_offset = 1;
-}
-
-// Method `CloseWriter`, asynchronous. Flushes, then drops the writer. It doesn't close the
-// Stream.
+// Method `CloseWriter`, asynchronous. Waits for a write in flight, then drops the writer. It
+// doesn't close the Stream. It fails with the writer's error if a write failed.
 message CloseWriterRequest {
     uint64 writer_id = 1;
 }
 
 message CloseWriterResponse {
+    // The offset of the newest Record the writer wrote. Zero if it wrote none.
     int64 last_offset = 1;
 }
 
-// Method `CloseWriters`, asynchronous. Flushes and drops every open writer, for example when the
-// plugin's client or Worker shuts down, so no flusher task stays behind.
+// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the plugin's
+// client or Worker shuts down.
 message CloseWritersRequest {
 }
 
 message CloseWritersResponse {
-    // The writers whose last flush failed.
+    // The writers that had failed.
     int32 failed = 1;
 }
 
diff --git a/scripts/_wit/stream-service.wit b/scripts/_wit/stream-service.wit
index bcf5885be..0b48fafcf 100644
--- a/scripts/_wit/stream-service.wit
+++ b/scripts/_wit/stream-service.wit
@@ -1,8 +1,8 @@
-// Draft. The Stream service as a System Nexus service, for the Brook v2 API proposal.
-// The records are backed by the messages in stream_service.proto (same directory), which
-// is a draft of temporal/api/streamservice/v1/. For the draft, every message and enum
-// is in one package. The spec carries no per-language names. Each language gets its
-// idiomatic shape from a hand-written facade over the generated code.
+// Draft. The Stream service as a System Nexus service, for the Brook API proposal. The records
+// are backed by the messages in ../protos/temporal/api/streamservice/v1/stream_service.proto.
+// This file is the source. Each SDK copies it and generates its bindings with nexgen. The spec
+// carries no per-language names. Each language gets its idiomatic shape from a hand-written
+// facade over the generated code.
 package temporal:nexus@1.0.0;
 
 world system {
@@ -33,11 +33,11 @@ interface stream-service {
   record stream-ref {
     /// @nexus.doc "The Stream id. 1 to 4096 bytes after encoding."
     id: string,
-    /// @nexus.doc "The Workflow that owns the Stream. Empty on a standalone, client or Activity ref."
+    /// @nexus.doc "The owning Workflow. Empty on a standalone, client or Activity ref."
     workflow-id: option<string>,
     /// @nexus.doc "The first run id of the owning run chain. Empty when workflow-id is."
     first-execution-run-id: option<string>,
-    /// @nexus.doc "The name of the backend that holds the Stream, as the Worker and client config register it. Empty is the default backend. The Store never reads it."
+    /// @nexus.doc "Registered backend name, empty for the default. The Store never reads it."
     backend: option<string>,
   }
 
@@ -123,9 +123,9 @@ interface stream-service {
     previous-completed-task: option<workflow-task-epoch>,
     /// @nexus.doc "Workflow only. Started event ids of failed or timed out Workflow Tasks."
     failed-tasks: list<s64>,
-    /// @nexus.doc "Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn."
+    /// @nexus.doc "Workflow retry only. The previous run, all of whose Records are withdrawn."
     withdraw-run: option<string>,
-    /// @nexus.doc "The run of the withdrawing Producer, so a Reader applies the Workflow rule without parsing ids. Empty for a client."
+    /// @nexus.doc "The withdrawing Producer's run, so no Reader parses ids. Empty for a client."
     run-id: option<string>,
   }
 
@@ -151,16 +151,16 @@ interface stream-service {
     close-failure: option<failure>,
   }
 
-  /// @nexus.doc "Create a Stream. Idempotent for one id, one Retention and one auto-close. ALREADY_EXISTS for another Retention or another auto-close."
+  /// @nexus.doc "Create a Stream idempotently. ALREADY_EXISTS for another Retention or auto-close."
   /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamRequest"
   record create-stream-request {
     /// @nexus.doc "Required from Workflow code. Empty elsewhere asks the service for an id."
     stream-id: string,
     /// @nexus.doc "The age after which a Record is dropped. Empty is the Store default."
     retention: option<duration>,
-    /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws."
+    /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws."
     withdraw-run: option<string>,
-    /// @nexus.doc "Close the Stream with no result when no Record was appended for this long. Empty never closes on its own. It is separate from Retention."
+    /// @nexus.doc "Close with no result after this long with no append. Empty never auto-closes."
     auto-close: option<duration>,
     /// @nexus.omit
     namespace: placeholder,
@@ -185,7 +185,8 @@ interface stream-service {
 
   // The meta key of each Stream has a format version v. Lua refuses a write to a Stream with a
   // higher v with FAILED_PRECONDITION and the detail stream_format_too_new.
-  /// @nexus.doc "Append one batch. FAILED_PRECONDITION if closed, fenced or stream_format_too_new. NOT_FOUND if absent."
+  /// @nexus.doc
+  ///   "Append. FAILED_PRECONDITION if closed, fenced or format too new. NOT_FOUND if absent."
   /// @nexus.proto "temporal.api.streamservice.v1.AppendRecordsRequest"
   record append-records-request {
     %stream: option<stream-ref>,
@@ -198,7 +199,7 @@ interface stream-service {
     previous-completed-task: option<workflow-task-epoch>,
     /// @nexus.doc "Workflow only. Failed or timed out tasks since this run's last flush here."
     failed-tasks: list<s64>,
-    /// @nexus.doc "Workflow only. A retry run's first batch or create: the previous run, whose Records it withdraws."
+    /// @nexus.doc "Retry run only. The previous run, whose Records this withdraws."
     withdraw-run: option<string>,
     /// @nexus.omit
     namespace: placeholder,
@@ -242,7 +243,7 @@ interface stream-service {
   /// @nexus.proto "temporal.api.streamservice.v1.ReadRecordsRequest"
   record read-records-request {
     %stream: option<stream-ref>,
-    /// @nexus.doc "Start at this offset. Empty uses start. INVALID_ARGUMENT if past the newest offset + 1."
+    /// @nexus.doc "Start offset. Empty uses start. INVALID_ARGUMENT past the newest offset + 1."
     /// @nexus.offset "stream-offset"
     from-offset: option<s64>,
     /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning."
@@ -292,7 +293,8 @@ interface stream-service {
   /// @nexus.operation name="CreateStream"
   create-stream: func(request: create-stream-request) -> create-stream-response;
 
-  /// @nexus.doc "In Workflow code, the Brook transport puts this call into the run's buffer, and the Workflow facade drops the response."
+  /// @nexus.doc
+  ///   "In Workflow code, Brook buffers this call in the run, and the facade drops the answer."
   /// @nexus.operation name="AppendRecords"
   append-records: func(request: append-records-request) -> append-records-response;
 
diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.pyi b/temporalio/api/streamservice/v1/stream_service_pb2.pyi
index 6bb44c320..1f50d9644 100644
--- a/temporalio/api/streamservice/v1/stream_service_pb2.pyi
+++ b/temporalio/api/streamservice/v1/stream_service_pb2.pyi
@@ -1,7 +1,7 @@
 """
 @generated by mypy-protobuf.  Do not edit manually!
 isort:skip_file
-Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 For the draft, every message and enum is in one package. A real change would split them
 into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 """
diff --git a/temporalio/api/streamservice/v1/stream_service_pb2_grpc.pyi b/temporalio/api/streamservice/v1/stream_service_pb2_grpc.pyi
index aa80dfa86..089ee084d 100644
--- a/temporalio/api/streamservice/v1/stream_service_pb2_grpc.pyi
+++ b/temporalio/api/streamservice/v1/stream_service_pb2_grpc.pyi
@@ -1,7 +1,7 @@
 """
 @generated by mypy-protobuf.  Do not edit manually!
 isort:skip_file
-Draft for the Brook v2 API proposal. It backs stream-service.wit in the same directory.
+Draft for the Brook API proposal. It backs crates/streams/nexus/stream-service.wit.
 For the draft, every message and enum is in one package. A real change would split them
 into temporal/api/stream/v1, temporal/api/enums/v1 and temporal/api/streamservice/v1.
 """
diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock
index 97bed84dd..c5ae55a30 100644
--- a/temporalio/bridge/Cargo.lock
+++ b/temporalio/bridge/Cargo.lock
@@ -2879,7 +2879,7 @@ dependencies = [
 [[package]]
 name = "temporalio-streams"
 version = "0.1.0"
-source = "git+https://github.com/moetemp/sdk-streams?rev=6254405544278746d98ac4bbb1f748ce174d1dc2#6254405544278746d98ac4bbb1f748ce174d1dc2"
+source = "git+https://github.com/moetemp/sdk-streams?rev=daeb5fc2a13e1f3457a461f109e7f5b8eb31951c#daeb5fc2a13e1f3457a461f109e7f5b8eb31951c"
 dependencies = [
  "async-trait",
  "aws-lc-rs",
diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml
index b5ab00c00..d2e25c4d3 100644
--- a/temporalio/bridge/Cargo.toml
+++ b/temporalio/bridge/Cargo.toml
@@ -94,7 +94,7 @@ url = "2.2"
 # crates, so the bridge sees one type of each.
 [dependencies.temporalio-streams]
 git = "https://github.com/moetemp/sdk-streams"
-rev = "6254405544278746d98ac4bbb1f748ce174d1dc2"
+rev = "daeb5fc2a13e1f3457a461f109e7f5b8eb31951c"
 default-features = false
 optional = true
 
diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py
index 7237ec3cd..089aa6a10 100644
--- a/temporalio/contrib/streams/__init__.py
+++ b/temporalio/contrib/streams/__init__.py
@@ -35,7 +35,6 @@
 )
 from temporalio.contrib.streams._errors import (
     ReadOnlyContextError,
-    StreamBufferFullError,
     StreamClosedError,
     StreamError,
     StreamExpiredError,
@@ -65,7 +64,6 @@
     "ReadOnlyContextError",
     "Record",
     "StreamBackend",
-    "StreamBufferFullError",
     "StreamClient",
     "StreamClosedError",
     "StreamDefinition",
diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py
index be0e2bcce..85d2323f9 100644
--- a/temporalio/contrib/streams/_errors.py
+++ b/temporalio/contrib/streams/_errors.py
@@ -73,10 +73,6 @@ class StreamOffsetError(StreamError):
     """The offset is past the newest one plus one, or Retention removed it."""
 
 
-class StreamBufferFullError(StreamError):
-    """The writer's buffer is full. Flush, or publish less often."""
-
-
 class StreamUnsupportedError(StreamError):
     """The Store lacks a capability that the call needs. The message ends with its name."""
 
@@ -96,7 +92,6 @@ class ReadOnlyContextError(StreamError):
     "stream_fenced": StreamFencedError,
     "stream_storage": StreamStorageError,
     "stream_offset": StreamOffsetError,
-    "stream_buffer_full": StreamBufferFullError,
     "stream_unsupported": StreamUnsupportedError,
 }
 
diff --git a/temporalio/contrib/streams/proto/v1/__init__.py b/temporalio/contrib/streams/proto/v1/__init__.py
index 6637363e7..c7e82df4f 100644
--- a/temporalio/contrib/streams/proto/v1/__init__.py
+++ b/temporalio/contrib/streams/proto/v1/__init__.py
@@ -9,12 +9,8 @@
     DrainClosesResponse,
     FlushRunBufferRequest,
     FlushRunBufferResponse,
-    FlushWriterRequest,
-    FlushWriterResponse,
     OpenWriterRequest,
     OpenWriterResponse,
-    PublishRequest,
-    PublishResponse,
     RedisStreamConfig,
     StreamErrorDetails,
     WorkflowCall,
@@ -33,12 +29,8 @@
     "DrainClosesResponse",
     "FlushRunBufferRequest",
     "FlushRunBufferResponse",
-    "FlushWriterRequest",
-    "FlushWriterResponse",
     "OpenWriterRequest",
     "OpenWriterResponse",
-    "PublishRequest",
-    "PublishResponse",
     "RedisStreamConfig",
     "StreamErrorDetails",
     "WorkflowCall",
diff --git a/temporalio/contrib/streams/proto/v1/extension_pb2.py b/temporalio/contrib/streams/proto/v1/extension_pb2.py
index 546cbb679..18252896e 100644
--- a/temporalio/contrib/streams/proto/v1/extension_pb2.py
+++ b/temporalio/contrib/streams/proto/v1/extension_pb2.py
@@ -21,7 +21,7 @@
 )
 
 DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
-    b'\n\'temporal/sdk/streams/v1/extension.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x32temporal/api/streamservice/v1/stream_service.proto"\xd6\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0e\x64urable_writes\x18\x06 \x01(\x05\x12\x32\n\x0f\x64urable_timeout\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t"\x91\x01\n\x1a\x42ufferWorkflowCallsRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x14\n\x0cwithdraw_run\x18\x02 \x01(\t\x12\x34\n\x05\x63\x61lls\x18\x03 \x03(\x0b\x32%.temporal.sdk.streams.v1.WorkflowCall\x12\x17\n\x0f\x63hain_continues\x18\x04 \x01(\x08"\x80\x02\n\x0cWorkflowCall\x12\x44\n\x06\x63reate\x18\x01 \x01(\x0b\x32\x32.temporal.api.streamservice.v1.CreateStreamRequestH\x00\x12\x45\n\x06\x61ppend\x18\x02 \x01(\x0b\x32\x33.temporal.api.streamservice.v1.AppendRecordsRequestH\x00\x12\x42\n\x05\x63lose\x18\x03 \x01(\x0b\x32\x31.temporal.api.streamservice.v1.CloseStreamRequestH\x00\x12\x17\n\x0fwhen_run_closes\x18\x04 \x01(\x08\x42\x06\n\x04\x63\x61ll"\x1d\n\x1b\x42ufferWorkflowCallsResponse"\xaf\x01\n\x15\x46lushRunBufferRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12G\n\rworkflow_task\x18\x03 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12*\n\x07timeout\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration")\n\x16\x46lushRunBufferResponse\x12\x0f\n\x07records\x18\x01 \x01(\x05"\xbb\x02\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x31\n\x0e\x62\x61tch_interval\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0emax_batch_size\x18\x05 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1c\n\x14max_buffered_records\x18\x07 \x01(\x05"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"v\n\x0ePublishRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12\x13\n\x0b\x66orce_flush\x18\x03 \x01(\x08"\x11\n\x0fPublishResponse"f\n\x13WriteRecordsRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord"A\n\x14WriteRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\'\n\x12\x46lushWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x46lushWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05"@\n\x12\x44rainClosesRequest\x12*\n\x07timeout\x18\x01 \x01(\x0b\x32\x19.google.protobuf.Duration"&\n\x13\x44rainClosesResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x05"$\n\x12StreamErrorDetails\x12\x0e\n\x06reason\x18\x01 \x01(\tb\x06proto3'
+    b'\n\'temporal/sdk/streams/v1/extension.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x32temporal/api/streamservice/v1/stream_service.proto"\xd6\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0e\x64urable_writes\x18\x06 \x01(\x05\x12\x32\n\x0f\x64urable_timeout\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t"\x91\x01\n\x1a\x42ufferWorkflowCallsRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x14\n\x0cwithdraw_run\x18\x02 \x01(\t\x12\x34\n\x05\x63\x61lls\x18\x03 \x03(\x0b\x32%.temporal.sdk.streams.v1.WorkflowCall\x12\x17\n\x0f\x63hain_continues\x18\x04 \x01(\x08"\x80\x02\n\x0cWorkflowCall\x12\x44\n\x06\x63reate\x18\x01 \x01(\x0b\x32\x32.temporal.api.streamservice.v1.CreateStreamRequestH\x00\x12\x45\n\x06\x61ppend\x18\x02 \x01(\x0b\x32\x33.temporal.api.streamservice.v1.AppendRecordsRequestH\x00\x12\x42\n\x05\x63lose\x18\x03 \x01(\x0b\x32\x31.temporal.api.streamservice.v1.CloseStreamRequestH\x00\x12\x17\n\x0fwhen_run_closes\x18\x04 \x01(\x08\x42\x06\n\x04\x63\x61ll"\x1d\n\x1b\x42ufferWorkflowCallsResponse"\xaf\x01\n\x15\x46lushRunBufferRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x0e\n\x06run_id\x18\x02 \x01(\t\x12G\n\rworkflow_task\x18\x03 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12*\n\x07timeout\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration")\n\x16\x46lushRunBufferResponse\x12\x0f\n\x07records\x18\x01 \x01(\x05"\x9c\x02\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x16\n\x0emax_batch_size\x18\x05 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x06 \x01(\x0b\x32\x19.google.protobuf.DurationJ\x04\x08\x04\x10\x05J\x04\x08\x07\x10\x08R\x0e\x62\x61tch_intervalR\x14max_buffered_records"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"f\n\x13WriteRecordsRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12<\n\x07records\x18\x02 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord"A\n\x14WriteRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05"@\n\x12\x44rainClosesRequest\x12*\n\x07timeout\x18\x01 \x01(\x0b\x32\x19.google.protobuf.Duration"&\n\x13\x44rainClosesResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x05"$\n\x12StreamErrorDetails\x12\x0e\n\x06reason\x18\x01 \x01(\tb\x06proto3'
 )
 
 
@@ -37,12 +37,8 @@
 _FLUSHRUNBUFFERRESPONSE = DESCRIPTOR.message_types_by_name["FlushRunBufferResponse"]
 _OPENWRITERREQUEST = DESCRIPTOR.message_types_by_name["OpenWriterRequest"]
 _OPENWRITERRESPONSE = DESCRIPTOR.message_types_by_name["OpenWriterResponse"]
-_PUBLISHREQUEST = DESCRIPTOR.message_types_by_name["PublishRequest"]
-_PUBLISHRESPONSE = DESCRIPTOR.message_types_by_name["PublishResponse"]
 _WRITERECORDSREQUEST = DESCRIPTOR.message_types_by_name["WriteRecordsRequest"]
 _WRITERECORDSRESPONSE = DESCRIPTOR.message_types_by_name["WriteRecordsResponse"]
-_FLUSHWRITERREQUEST = DESCRIPTOR.message_types_by_name["FlushWriterRequest"]
-_FLUSHWRITERRESPONSE = DESCRIPTOR.message_types_by_name["FlushWriterResponse"]
 _CLOSEWRITERREQUEST = DESCRIPTOR.message_types_by_name["CloseWriterRequest"]
 _CLOSEWRITERRESPONSE = DESCRIPTOR.message_types_by_name["CloseWriterResponse"]
 _CLOSEWRITERSREQUEST = DESCRIPTOR.message_types_by_name["CloseWritersRequest"]
@@ -138,28 +134,6 @@
 )
 _sym_db.RegisterMessage(OpenWriterResponse)
 
-PublishRequest = _reflection.GeneratedProtocolMessageType(
-    "PublishRequest",
-    (_message.Message,),
-    {
-        "DESCRIPTOR": _PUBLISHREQUEST,
-        "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2",
-        # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.PublishRequest)
-    },
-)
-_sym_db.RegisterMessage(PublishRequest)
-
-PublishResponse = _reflection.GeneratedProtocolMessageType(
-    "PublishResponse",
-    (_message.Message,),
-    {
-        "DESCRIPTOR": _PUBLISHRESPONSE,
-        "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2",
-        # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.PublishResponse)
-    },
-)
-_sym_db.RegisterMessage(PublishResponse)
-
 WriteRecordsRequest = _reflection.GeneratedProtocolMessageType(
     "WriteRecordsRequest",
     (_message.Message,),
@@ -182,28 +156,6 @@
 )
 _sym_db.RegisterMessage(WriteRecordsResponse)
 
-FlushWriterRequest = _reflection.GeneratedProtocolMessageType(
-    "FlushWriterRequest",
-    (_message.Message,),
-    {
-        "DESCRIPTOR": _FLUSHWRITERREQUEST,
-        "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2",
-        # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FlushWriterRequest)
-    },
-)
-_sym_db.RegisterMessage(FlushWriterRequest)
-
-FlushWriterResponse = _reflection.GeneratedProtocolMessageType(
-    "FlushWriterResponse",
-    (_message.Message,),
-    {
-        "DESCRIPTOR": _FLUSHWRITERRESPONSE,
-        "__module__": "temporalio.contrib.streams.proto.v1.extension_pb2",
-        # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FlushWriterResponse)
-    },
-)
-_sym_db.RegisterMessage(FlushWriterResponse)
-
 CloseWriterRequest = _reflection.GeneratedProtocolMessageType(
     "CloseWriterRequest",
     (_message.Message,),
@@ -296,33 +248,25 @@
     _FLUSHRUNBUFFERRESPONSE._serialized_start = 1113
     _FLUSHRUNBUFFERRESPONSE._serialized_end = 1154
     _OPENWRITERREQUEST._serialized_start = 1157
-    _OPENWRITERREQUEST._serialized_end = 1472
-    _OPENWRITERRESPONSE._serialized_start = 1474
-    _OPENWRITERRESPONSE._serialized_end = 1513
-    _PUBLISHREQUEST._serialized_start = 1515
-    _PUBLISHREQUEST._serialized_end = 1633
-    _PUBLISHRESPONSE._serialized_start = 1635
-    _PUBLISHRESPONSE._serialized_end = 1652
-    _WRITERECORDSREQUEST._serialized_start = 1654
-    _WRITERECORDSREQUEST._serialized_end = 1756
-    _WRITERECORDSRESPONSE._serialized_start = 1758
-    _WRITERECORDSRESPONSE._serialized_end = 1823
-    _FLUSHWRITERREQUEST._serialized_start = 1825
-    _FLUSHWRITERREQUEST._serialized_end = 1864
-    _FLUSHWRITERRESPONSE._serialized_start = 1866
-    _FLUSHWRITERRESPONSE._serialized_end = 1908
-    _CLOSEWRITERREQUEST._serialized_start = 1910
-    _CLOSEWRITERREQUEST._serialized_end = 1949
-    _CLOSEWRITERRESPONSE._serialized_start = 1951
-    _CLOSEWRITERRESPONSE._serialized_end = 1993
-    _CLOSEWRITERSREQUEST._serialized_start = 1995
-    _CLOSEWRITERSREQUEST._serialized_end = 2016
-    _CLOSEWRITERSRESPONSE._serialized_start = 2018
-    _CLOSEWRITERSRESPONSE._serialized_end = 2056
-    _DRAINCLOSESREQUEST._serialized_start = 2058
-    _DRAINCLOSESREQUEST._serialized_end = 2122
-    _DRAINCLOSESRESPONSE._serialized_start = 2124
-    _DRAINCLOSESRESPONSE._serialized_end = 2162
-    _STREAMERRORDETAILS._serialized_start = 2164
-    _STREAMERRORDETAILS._serialized_end = 2200
+    _OPENWRITERREQUEST._serialized_end = 1441
+    _OPENWRITERRESPONSE._serialized_start = 1443
+    _OPENWRITERRESPONSE._serialized_end = 1482
+    _WRITERECORDSREQUEST._serialized_start = 1484
+    _WRITERECORDSREQUEST._serialized_end = 1586
+    _WRITERECORDSRESPONSE._serialized_start = 1588
+    _WRITERECORDSRESPONSE._serialized_end = 1653
+    _CLOSEWRITERREQUEST._serialized_start = 1655
+    _CLOSEWRITERREQUEST._serialized_end = 1694
+    _CLOSEWRITERRESPONSE._serialized_start = 1696
+    _CLOSEWRITERRESPONSE._serialized_end = 1738
+    _CLOSEWRITERSREQUEST._serialized_start = 1740
+    _CLOSEWRITERSREQUEST._serialized_end = 1761
+    _CLOSEWRITERSRESPONSE._serialized_start = 1763
+    _CLOSEWRITERSRESPONSE._serialized_end = 1801
+    _DRAINCLOSESREQUEST._serialized_start = 1803
+    _DRAINCLOSESREQUEST._serialized_end = 1867
+    _DRAINCLOSESRESPONSE._serialized_start = 1869
+    _DRAINCLOSESRESPONSE._serialized_end = 1907
+    _STREAMERRORDETAILS._serialized_start = 1909
+    _STREAMERRORDETAILS._serialized_end = 1945
 # @@protoc_insertion_point(module_scope)
diff --git a/temporalio/contrib/streams/proto/v1/extension_pb2.pyi b/temporalio/contrib/streams/proto/v1/extension_pb2.pyi
index fad30118e..72bbb440d 100644
--- a/temporalio/contrib/streams/proto/v1/extension_pb2.pyi
+++ b/temporalio/contrib/streams/proto/v1/extension_pb2.pyi
@@ -324,17 +324,18 @@ class FlushRunBufferResponse(google.protobuf.message.Message):
 global___FlushRunBufferResponse = FlushRunBufferResponse
 
 class OpenWriterRequest(google.protobuf.message.Message):
-    """Method `OpenWriter`, asynchronous. A buffered writer for an Activity or a client Producer."""
+    """Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps
+    its epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty
+    batch, so the new epoch withdraws the earlier attempt's Records.
+    """
 
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
 
     NAMESPACE_FIELD_NUMBER: builtins.int
     STREAM_FIELD_NUMBER: builtins.int
     PRODUCER_FIELD_NUMBER: builtins.int
-    BATCH_INTERVAL_FIELD_NUMBER: builtins.int
     MAX_BATCH_SIZE_FIELD_NUMBER: builtins.int
     MAX_RETRY_DURATION_FIELD_NUMBER: builtins.int
-    MAX_BUFFERED_RECORDS_FIELD_NUMBER: builtins.int
     namespace: builtins.str
     @property
     def stream(
@@ -344,16 +345,11 @@ class OpenWriterRequest(google.protobuf.message.Message):
     def producer(
         self,
     ) -> temporalio.api.streamservice.v1.stream_service_pb2.Producer: ...
-    @property
-    def batch_interval(self) -> google.protobuf.duration_pb2.Duration:
-        """How often the writer flushes. Empty is 100 ms."""
     max_batch_size: builtins.int
-    """A flush sends at most this many Records in one batch. Zero is 1000."""
+    """A write sends at most this many Records in one batch. Zero is 1000."""
     @property
     def max_retry_duration(self) -> google.protobuf.duration_pb2.Duration:
-        """How long a flush retries a Store failure. Empty is 10 s."""
-    max_buffered_records: builtins.int
-    """A publish past this many buffered Records fails with `stream_buffer_full`. Zero is 10000."""
+        """How long a write retries a Store failure. Empty is 10 s."""
     def __init__(
         self,
         *,
@@ -362,16 +358,12 @@ class OpenWriterRequest(google.protobuf.message.Message):
         | None = ...,
         producer: temporalio.api.streamservice.v1.stream_service_pb2.Producer
         | None = ...,
-        batch_interval: google.protobuf.duration_pb2.Duration | None = ...,
         max_batch_size: builtins.int = ...,
         max_retry_duration: google.protobuf.duration_pb2.Duration | None = ...,
-        max_buffered_records: builtins.int = ...,
     ) -> None: ...
     def HasField(
         self,
         field_name: typing_extensions.Literal[
-            "batch_interval",
-            b"batch_interval",
             "max_retry_duration",
             b"max_retry_duration",
             "producer",
@@ -383,12 +375,8 @@ class OpenWriterRequest(google.protobuf.message.Message):
     def ClearField(
         self,
         field_name: typing_extensions.Literal[
-            "batch_interval",
-            b"batch_interval",
             "max_batch_size",
             b"max_batch_size",
-            "max_buffered_records",
-            b"max_buffered_records",
             "max_retry_duration",
             b"max_retry_duration",
             "namespace",
@@ -418,63 +406,12 @@ class OpenWriterResponse(google.protobuf.message.Message):
 
 global___OpenWriterResponse = OpenWriterResponse
 
-class PublishRequest(google.protobuf.message.Message):
-    """Method `Publish`, synchronous. Adds encoded Records to a writer's buffer."""
-
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-    WRITER_ID_FIELD_NUMBER: builtins.int
-    RECORDS_FIELD_NUMBER: builtins.int
-    FORCE_FLUSH_FIELD_NUMBER: builtins.int
-    writer_id: builtins.int
-    @property
-    def records(
-        self,
-    ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
-        temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord
-    ]: ...
-    force_flush: builtins.bool
-    """Start a flush at once."""
-    def __init__(
-        self,
-        *,
-        writer_id: builtins.int = ...,
-        records: collections.abc.Iterable[
-            temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord
-        ]
-        | None = ...,
-        force_flush: builtins.bool = ...,
-    ) -> None: ...
-    def ClearField(
-        self,
-        field_name: typing_extensions.Literal[
-            "force_flush",
-            b"force_flush",
-            "records",
-            b"records",
-            "writer_id",
-            b"writer_id",
-        ],
-    ) -> None: ...
-
-global___PublishRequest = PublishRequest
-
-class PublishResponse(google.protobuf.message.Message):
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-    def __init__(
-        self,
-    ) -> None: ...
-
-global___PublishResponse = PublishResponse
-
 class WriteRecordsRequest(google.protobuf.message.Message):
-    """Method `WriteRecords`, asynchronous. Appends Records at once for an open writer, after the
-    Records that `Publish` buffered, and answers when they are in the Store. It keeps the writer's
-    Producer epoch and `sequence` order, and a Store failure resends the same batch, so it goes in
-    once. Records past the Store's batch limit go as more batches, in order. An awaited publish of
-    an Activity or a client uses this call, so it needs no interval flush. A failure after the
-    Store retries fails the writer, as a failed flush does.
+    """Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and answers
+    when they are in the Store. It keeps the writer's Producer epoch and `sequence` order, and a
+    Store failure resends the same batch, so it goes in once. Records past the Store's batch limit
+    go as more batches, in order. An awaited publish of an Activity or a client uses this call. A
+    failure after the Store retries fails the writer, since the batch may be in the Store.
     """
 
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -529,44 +466,9 @@ class WriteRecordsResponse(google.protobuf.message.Message):
 
 global___WriteRecordsResponse = WriteRecordsResponse
 
-class FlushWriterRequest(google.protobuf.message.Message):
-    """Method `FlushWriter`, asynchronous. Waits until the Records published so far are in the Store."""
-
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-    WRITER_ID_FIELD_NUMBER: builtins.int
-    writer_id: builtins.int
-    def __init__(
-        self,
-        *,
-        writer_id: builtins.int = ...,
-    ) -> None: ...
-    def ClearField(
-        self, field_name: typing_extensions.Literal["writer_id", b"writer_id"]
-    ) -> None: ...
-
-global___FlushWriterRequest = FlushWriterRequest
-
-class FlushWriterResponse(google.protobuf.message.Message):
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-    LAST_OFFSET_FIELD_NUMBER: builtins.int
-    last_offset: builtins.int
-    """The offset of the newest Record the writer wrote. Zero if it wrote none."""
-    def __init__(
-        self,
-        *,
-        last_offset: builtins.int = ...,
-    ) -> None: ...
-    def ClearField(
-        self, field_name: typing_extensions.Literal["last_offset", b"last_offset"]
-    ) -> None: ...
-
-global___FlushWriterResponse = FlushWriterResponse
-
 class CloseWriterRequest(google.protobuf.message.Message):
-    """Method `CloseWriter`, asynchronous. Flushes, then drops the writer. It doesn't close the
-    Stream.
+    """Method `CloseWriter`, asynchronous. Waits for a write in flight, then drops the writer. It
+    doesn't close the Stream. It fails with the writer's error if a write failed.
     """
 
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -589,6 +491,7 @@ class CloseWriterResponse(google.protobuf.message.Message):
 
     LAST_OFFSET_FIELD_NUMBER: builtins.int
     last_offset: builtins.int
+    """The offset of the newest Record the writer wrote. Zero if it wrote none."""
     def __init__(
         self,
         *,
@@ -601,8 +504,8 @@ class CloseWriterResponse(google.protobuf.message.Message):
 global___CloseWriterResponse = CloseWriterResponse
 
 class CloseWritersRequest(google.protobuf.message.Message):
-    """Method `CloseWriters`, asynchronous. Flushes and drops every open writer, for example when the
-    plugin's client or Worker shuts down, so no flusher task stays behind.
+    """Method `CloseWriters`, asynchronous. Drops every open writer, for example when the plugin's
+    client or Worker shuts down.
     """
 
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -618,7 +521,7 @@ class CloseWritersResponse(google.protobuf.message.Message):
 
     FAILED_FIELD_NUMBER: builtins.int
     failed: builtins.int
-    """The writers whose last flush failed."""
+    """The writers that had failed."""
     def __init__(
         self,
         *,
diff --git a/temporalio/contrib/streams/system/client.py b/temporalio/contrib/streams/system/client.py
index 8ecfc6238..2883b3f1c 100644
--- a/temporalio/contrib/streams/system/client.py
+++ b/temporalio/contrib/streams/system/client.py
@@ -169,8 +169,8 @@ async def create_stream(self, request: CreateStreamRequest) -> CreateStreamRespo
     async def append_records(
         self, request: AppendRecordsRequest
     ) -> AppendRecordsResponse:
-        """In Workflow code, the Brook transport puts this call into the run's buffer, and
-        the Workflow facade drops the response.
+        """In Workflow code, Brook buffers this call in the run, and the facade drops the
+        answer.
         """
         payload = _proto_payload(
             _AppendRecordsRequestTransferTypeConverter().to_transfer_type(request)
diff --git a/temporalio/contrib/streams/system/models.py b/temporalio/contrib/streams/system/models.py
index 23b805dc7..2304fc581 100644
--- a/temporalio/contrib/streams/system/models.py
+++ b/temporalio/contrib/streams/system/models.py
@@ -23,9 +23,7 @@
 
 @dataclasses.dataclass(slots=True)
 class CreateStreamRequest:
-    """Create a Stream. Idempotent for one id, one Retention and one auto-close.
-    ALREADY_EXISTS for another Retention or another auto-close.
-    """
+    """Create a Stream idempotently. ALREADY_EXISTS for another Retention or auto-close."""
 
     stream_id: str
     retention: datetime.timedelta | None = None
@@ -213,8 +211,8 @@ def to_transfer_type(
 
 @dataclasses.dataclass(slots=True)
 class AppendRecordsRequest:
-    """Append one batch. FAILED_PRECONDITION if closed, fenced or stream_format_too_new.
-    NOT_FOUND if absent.
+    """Append. FAILED_PRECONDITION if closed, fenced or format too new. NOT_FOUND if
+    absent.
     """
 
     stream: StreamRef | None = None
diff --git a/temporalio/contrib/streams/system/operations/append_records.py b/temporalio/contrib/streams/system/operations/append_records.py
index 0d4a1a574..43324b0ee 100644
--- a/temporalio/contrib/streams/system/operations/append_records.py
+++ b/temporalio/contrib/streams/system/operations/append_records.py
@@ -38,8 +38,8 @@ async def append_records(
     failed_tasks: list[int] | None = None,
     withdraw_run: str | None = None,
 ) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]:
-    """In Workflow code, the Brook transport puts this call into the run's buffer, and the
-    Workflow facade drops the response.
+    """In Workflow code, Brook buffers this call in the run, and the facade drops the
+    answer.
 
     Args:
         producer: Required outside a Workflow. In a Workflow the transport sets it.
@@ -49,8 +49,7 @@ async def append_records(
             the run.
         failed_tasks: Workflow only. Failed or timed out tasks since this run's last
             flush here.
-        withdraw_run: Workflow only. A retry run's first batch or create: the previous
-            run, whose Records it withdraws.
+        withdraw_run: Retry run only. The previous run, whose Records this withdraws.
     """
     request = AppendRecordsRequest(
         stream=stream,
diff --git a/temporalio/contrib/streams/system/operations/create_stream.py b/temporalio/contrib/streams/system/operations/create_stream.py
index b798ea98e..2135918aa 100644
--- a/temporalio/contrib/streams/system/operations/create_stream.py
+++ b/temporalio/contrib/streams/system/operations/create_stream.py
@@ -38,10 +38,9 @@ async def create_stream(
         stream_id: Required from Workflow code. Empty elsewhere asks the service for an
             id.
         retention: The age after which a Record is dropped. Empty is the Store default.
-        withdraw_run: Workflow only. A retry run's first batch or create: the previous
-            run, whose Records it withdraws.
-        auto_close: Close the Stream with no result when no Record was appended for this
-            long. Empty never closes on its own. It is separate from Retention.
+        withdraw_run: Retry run only. The previous run, whose Records this withdraws.
+        auto_close: Close with no result after this long with no append. Empty never
+            auto-closes.
     """
     request = CreateStreamRequest(
         stream_id=stream_id,
diff --git a/temporalio/contrib/streams/system/operations/read_records.py b/temporalio/contrib/streams/system/operations/read_records.py
index 8e5675994..2bfce568d 100644
--- a/temporalio/contrib/streams/system/operations/read_records.py
+++ b/temporalio/contrib/streams/system/operations/read_records.py
@@ -43,8 +43,8 @@ async def read_records(
     """Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream.
 
     Args:
-        from_offset: Start at this offset. Empty uses start. INVALID_ARGUMENT if past
-            the newest offset + 1.
+        from_offset: Start offset. Empty uses start. INVALID_ARGUMENT past the newest
+            offset + 1.
         start: Where to start when from-offset is empty. Unspecified is the beginning.
         topics: Topics to read, by exact match. With topic-prefixes, empty reads every
             Topic.
diff --git a/tests/contrib/streams/fake_extension.py b/tests/contrib/streams/fake_extension.py
index 9704944a5..695982ecf 100644
--- a/tests/contrib/streams/fake_extension.py
+++ b/tests/contrib/streams/fake_extension.py
@@ -47,20 +47,6 @@ def _append(self, stream_id: str, record: stream_pb.StreamRecord) -> int:
         return record.offset
 
     def call(self, method: str, input: bytes) -> bytes:
-        if method == "Publish":
-            request = ext_pb.PublishRequest.FromString(input)
-            stream_id, producer = self.writers[request.writer_id]
-            for record in request.records:
-                self._append(
-                    stream_id,
-                    stream_pb.StreamRecord(
-                        kind=stream_pb.StreamRecordKind.STREAM_RECORD_KIND_DATA,
-                        topic=record.topic,
-                        body=record.body,
-                        producer=producer,
-                    ),
-                )
-            return b""
         raise _error(RPCStatusCode.UNIMPLEMENTED, method)
 
     async def call_async(self, method: str, input: bytes) -> bytes:
@@ -95,8 +81,8 @@ async def call_async(self, method: str, input: bytes) -> bytes:
                 first_offset=offsets[0] if offsets else 0,
                 last_offset=offsets[-1] if offsets else 0,
             ).SerializeToString()
-        if method in ("FlushWriter", "CloseWriter"):
-            return ext_pb.FlushWriterResponse(
+        if method == "CloseWriter":
+            return ext_pb.CloseWriterResponse(
                 last_offset=self.next_offset - 1
             ).SerializeToString()
         if method == "CloseStream":