From e2eca25d3066abce5f72e567644fcca2237a4057 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 01:31:48 -0700
Subject: [PATCH 1/6] Added Workflow reads with subscribe on both backends.

---
 temporalio/contrib/streams/_access.py         |  16 +-
 temporalio/contrib/streams/_errors.py         |  18 ++
 .../contrib/streams/_workflow_reader.py       | 182 ++++++++++++
 temporalio/contrib/streams/native.py          |  26 +-
 temporalio/contrib/streams/redis.py           |  79 ++++-
 temporalio/contrib/streams/workflow.py        |  77 ++++-
 tests/contrib/streams/conftest.py             |  19 ++
 tests/contrib/streams/fake_native.py          |  40 ++-
 tests/contrib/streams/test_facade.py          |   6 +-
 tests/contrib/streams/test_native_server.py   | 101 +++++++
 tests/contrib/streams/test_workflow_reads.py  | 275 ++++++++++++++++++
 11 files changed, 824 insertions(+), 15 deletions(-)
 create mode 100644 temporalio/contrib/streams/_workflow_reader.py
 create mode 100644 tests/contrib/streams/test_native_server.py
 create mode 100644 tests/contrib/streams/test_workflow_reads.py

diff --git a/temporalio/contrib/streams/_access.py b/temporalio/contrib/streams/_access.py
index 0ce160ecf..94a89f8e4 100644
--- a/temporalio/contrib/streams/_access.py
+++ b/temporalio/contrib/streams/_access.py
@@ -116,22 +116,26 @@ async def close(self, result: Any | None, failure: BaseException | None) -> None
 
 def read_spec(
     handle: StreamHandle[Any],
+    # Any is a Topic of Workflow code, which this module can't import.
     topics: Sequence[
-        str | TopicDefinition[Any] | _AsyncTopic[Any] | SyncTopicBase[Any]
+        str | TopicDefinition[Any] | _AsyncTopic[Any] | SyncTopicBase[Any] | Any
     ],
 ) -> tuple[list[str] | None, dict[str, Any]]:
-    """The Topic names of a read and the decode type of each known Topic."""
+    """The Topic names of a read and the decode type of each known Topic.
+
+    A Topic object of any context, such as a ``WorkflowTopic``, gives its name and type.
+    """
     types = handle._known_types()
     names = []
     for topic in topics:
-        if isinstance(topic, (_AsyncTopic, SyncTopicBase)):
-            name, topic_type = topic.name, topic.type
-        elif isinstance(topic, TopicDefinition):
+        if isinstance(topic, TopicDefinition):
             name, topic_type = handle._topic_spec(topic, None)
-        else:
+        elif isinstance(topic, str):
             # A read names a Topic outside the definition with no type, and its values
             # decode as the converter gives them.
             name, topic_type = topic, types.get(topic)
+        else:
+            name, topic_type = topic.name, topic.type
         names.append(name)
         if topic_type is not None:
             types[name] = topic_type
diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py
index 85d2323f9..2ad3ebc6f 100644
--- a/temporalio/contrib/streams/_errors.py
+++ b/temporalio/contrib/streams/_errors.py
@@ -104,3 +104,21 @@ def from_extension_error(err: temporalio.worker.ExtensionError) -> StreamError:
     return _BY_REASON.get(reason, StreamError)(
         err.message, reason=reason, code=err.code
     )
+
+
+_ERROR_CLASSES: list[type[StreamError]] = [StreamError, *_BY_REASON.values()]
+_BY_NAME: dict[str, type[StreamError]] = {cls.__name__: cls for cls in _ERROR_CLASSES}
+
+
+def from_application_error(
+    err: temporalio.exceptions.ApplicationError,
+) -> StreamError | None:
+    """The Stream error that an application error of a failed call carries, if any.
+
+    A Stream error's type is its class name, and its first detail is its reason.
+    """
+    cls = _BY_NAME.get(err.type or "")
+    if cls is None:
+        return None
+    reason = err.details[0] if err.details and isinstance(err.details[0], str) else ""
+    return cls(err.message, reason=reason)
diff --git a/temporalio/contrib/streams/_workflow_reader.py b/temporalio/contrib/streams/_workflow_reader.py
new file mode 100644
index 000000000..22a24bcf7
--- /dev/null
+++ b/temporalio/contrib/streams/_workflow_reader.py
@@ -0,0 +1,182 @@
+"""Reads of a Stream in Workflow code.
+
+Each page is a call whose answer is in History, so a replay reads the same items. A Stream
+plugin's outbound interceptor reads a page with ``_temporal_stream_read_page``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator, Mapping, Sequence
+from datetime import datetime, timedelta
+from typing import Any
+
+import temporalio.api.streamservice.v1 as stream_pb
+import temporalio.common
+import temporalio.converter
+import temporalio.exceptions
+import temporalio.workflow
+from temporalio.contrib.streams import _convert, _errors
+from temporalio.contrib.streams._items import Closed, Record, Superseded
+from temporalio.contrib.streams.system import models
+
+Item = Record[Any] | Superseded | Closed
+
+# The service waits at most this long for a Stream that nobody created yet, per call.
+_MAX_CREATE_WAIT = timedelta(seconds=60)
+# A backend that can't park a read pauses on a durable timer after a page with no Record,
+# from the first pause up to the last.
+_FIRST_PAUSE = timedelta(seconds=1)
+_LAST_PAUSE = timedelta(seconds=60)
+
+
+def reader_of(backend: str) -> Any:
+    """The outbound interceptor of the Stream plugin with this backend name."""
+    outbound = getattr(temporalio.workflow._Runtime.current(), "_outbound", None)
+    while outbound is not None:
+        if (
+            getattr(outbound, "_temporal_stream_transport", False)
+            and getattr(outbound, "_temporal_stream_backend_name", None) == backend
+        ):
+            return outbound
+        outbound = getattr(outbound, "next", None)
+    raise RuntimeError(f"No Stream plugin of the Worker has the backend {backend}")
+
+
+async def read(
+    reader: Any,
+    ref: stream_pb.StreamRef,
+    *,
+    topics: Sequence[str] | None,
+    prefixes: Sequence[str],
+    types: Mapping[str, Any],
+    from_offset: int | None,
+    wait_new_records: bool,
+    create_wait: timedelta | None,
+) -> AsyncGenerator[Item, None]:
+    """Give the items of a Stream in order, until a close, or the newest Record when
+    ``wait_new_records`` is False.
+    """
+    request = stream_pb.ReadRecordsRequest(
+        stream=ref, topics=list(topics or []), topic_prefixes=list(prefixes)
+    )
+    if from_offset:
+        request.from_offset = from_offset
+    else:
+        request.start = stream_pb.ReadStart.READ_START_BEGINNING
+    create_until = temporalio.workflow.now() + create_wait if create_wait else None
+    pause = _FIRST_PAUSE
+    while True:
+        if temporalio.workflow.unsafe.is_read_only():
+            raise _errors.ReadOnlyContextError("Can't read a Stream in read-only code")
+        page_request = stream_pb.ReadRecordsRequest()
+        page_request.CopyFrom(request)
+        if wait_new_records:
+            page_request.wait.FromTimedelta(reader._temporal_stream_read_wait)
+        left = _left(create_until)
+        if left:
+            page_request.create_wait.FromTimedelta(min(left, _MAX_CREATE_WAIT))
+        try:
+            page = await reader._temporal_stream_read_page(page_request)
+        except _errors.StreamNotFoundError:
+            if not _left(create_until):
+                raise
+            await temporalio.workflow.sleep(pause)
+            pause = min(pause * 2, _LAST_PAUSE)
+            continue
+        for record in page.records:
+            item = workflow_item(record, types)
+            if item is None:
+                continue
+            yield item
+            if isinstance(item, Closed):
+                return
+        if page.done:
+            return
+        if page.next_offset:
+            request.from_offset = page.next_offset
+            request.ClearField("start")
+        if page.records:
+            pause = _FIRST_PAUSE
+            continue
+        if not wait_new_records:
+            return
+        if not reader._temporal_stream_parks_reads:
+            # The server wake comes later. Until then a quiet Stream costs one timer and one
+            # read per pause.
+            await temporalio.workflow.sleep(pause)
+            pause = min(pause * 2, _LAST_PAUSE)
+
+
+def _left(until: datetime | None) -> timedelta | None:
+    if until is None:
+        return None
+    left = until - temporalio.workflow.now()
+    return left if left > timedelta() else None
+
+
+def workflow_item(
+    record: stream_pb.StreamRecord, types: Mapping[str, Any]
+) -> Item | None:
+    """One read item, with the body decoded by the Workflow's converter and its Topic's type.
+
+    The codec already ran outside the sandbox, on the page.
+    """
+    converter = temporalio.workflow.payload_converter()
+    # The generated converter makes the producer and the supersession models, and the body
+    # stays a Payload that the Topic's type decodes.
+    with _convert.raw_bodies(_failure_converter()):
+        model = models._StreamRecordTransferTypeConverter().from_transfer_type(
+            record, models.StreamRecord
+        )
+    offset = model.offset or 0
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_DATA:
+        topic = model.topic or ""
+        return Record(
+            topic=topic,
+            data=_decode(converter, model.body, types.get(topic)),
+            offset=offset,
+            producer=model.producer,
+        )
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_SUPERSEDED:
+        if model.supersession is None:
+            return None
+        return Superseded(offset=offset, supersession=model.supersession)
+    if model.kind == stream_pb.StreamRecordKind.STREAM_RECORD_KIND_CLOSE:
+        return Closed(
+            offset=offset,
+            result=_decode(converter, model.close_result, None),
+            failure=model.close_failure,
+        )
+    return None
+
+
+def _decode(
+    converter: temporalio.converter.PayloadConverter, body: Any, type_hint: Any
+) -> Any:
+    if not isinstance(body, temporalio.common.RawValue):
+        return body
+    return converter.from_payload(body.payload, type_hint)
+
+
+def _failure_converter() -> temporalio.converter.FailureConverter:
+    runtime = temporalio.workflow._Runtime.current()
+    return getattr(
+        runtime,
+        "_workflow_context_failure_converter",
+        temporalio.converter.DefaultFailureConverter(),
+    )
+
+
+def error_of(err: BaseException) -> BaseException:
+    """The Stream error that a failed page carries as its cause, or the error itself.
+
+    A page's failure is in History, so a replay raises the same error.
+    """
+    cause: BaseException | None = err
+    while cause is not None:
+        if isinstance(cause, temporalio.exceptions.ApplicationError):
+            stream_error = _errors.from_application_error(cause)
+            if stream_error is not None:
+                return stream_error
+        cause = cause.__cause__
+    return err
diff --git a/temporalio/contrib/streams/native.py b/temporalio/contrib/streams/native.py
index 7e000f60e..cfc258df3 100644
--- a/temporalio/contrib/streams/native.py
+++ b/temporalio/contrib/streams/native.py
@@ -10,6 +10,7 @@
 import dataclasses
 import itertools
 from collections.abc import Coroutine, Generator
+from datetime import timedelta
 from typing import Any, NoReturn, TypeVar
 
 import google.protobuf.message
@@ -23,7 +24,7 @@
 import temporalio.service
 import temporalio.worker
 import temporalio.workflow
-from temporalio.contrib.streams import _activity, _errors
+from temporalio.contrib.streams import _activity, _errors, _workflow_reader
 from temporalio.contrib.streams import workflow as streams_workflow
 from temporalio.contrib.streams._backend import _REQUESTS, _RESPONSES, StreamBackend
 from temporalio.contrib.streams.proto import v1 as ext_pb
@@ -49,6 +50,9 @@
 
 _M = TypeVar("_M", bound=google.protobuf.message.Message)
 
+# How long a Workflow read asks the server to park when the Stream has no new Record. The
+# operation is async, so the wait costs no Worker resource.
+_READ_WAIT = timedelta(seconds=60)
 # The writer ids of this process. A native writer lives only in the process.
 _WRITER_IDS = itertools.count(1)
 
@@ -405,6 +409,10 @@ def _operation_input(
 
 class _NativeWorkflowOutbound(temporalio.worker.WorkflowOutboundInterceptor):
     _temporal_stream_transport = True
+    # The server parks a read at the end of the Stream, and completes it on an append or a
+    # close, so a reader needs no timer.
+    _temporal_stream_parks_reads = True
+    _temporal_stream_read_wait = _READ_WAIT
 
     def __init__(
         self,
@@ -440,6 +448,22 @@ async def start_system_nexus_operation(
         _schedule_now(_call_and_drop(super().start_system_nexus_operation, input))
         return _DoneHandle(result)
 
+    async def _temporal_stream_read_page(
+        self, request: stream_pb.ReadRecordsRequest
+    ) -> stream_pb.ReadRecordsResponse:
+        # Another Stream plugin further down the chain passes a call of this backend on.
+        token = streams_workflow._target.set(self._temporal_stream_backend_name)
+        try:
+            handle = await self.next.start_system_nexus_operation(
+                _operation_input("ReadRecords", request, stream_pb.ReadRecordsResponse)
+            )
+        finally:
+            streams_workflow._target.reset(token)
+        try:
+            return await handle
+        except temporalio.exceptions.NexusOperationError as err:
+            raise _workflow_reader.error_of(err) from None
+
     def continue_as_new(self, input: temporalio.worker.ContinueAsNewInput) -> NoReturn:
         open_streams = {
             stream_id: run_stream.backend
diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py
index b1ed0784b..ddafff2c6 100644
--- a/temporalio/contrib/streams/redis.py
+++ b/temporalio/contrib/streams/redis.py
@@ -24,7 +24,7 @@
 import temporalio.runtime
 import temporalio.worker
 import temporalio.workflow
-from temporalio.contrib.streams import _activity, _errors, _ids
+from temporalio.contrib.streams import _activity, _errors, _ids, _workflow_reader
 from temporalio.contrib.streams import workflow as streams_workflow
 from temporalio.contrib.streams._backend import StreamBackend
 from temporalio.contrib.streams._handle import StreamHandle
@@ -40,6 +40,9 @@
 _SERVICE = "temporal.api.streamservice.v1.StreamService"
 # How long a Worker's shutdown waits for the Stream closes that are still going out.
 _DRAIN_TIMEOUT = timedelta(seconds=10)
+# How long a Workflow read's Local Activity waits for new Records. Its timeout adds a margin.
+_READ_WAIT = timedelta(seconds=10)
+_READ_MARGIN = timedelta(seconds=10)
 # The plugin puts this key into the headers of each Activity and Local Activity that a Workflow
 # schedules. It keeps the Producer id the same across retries.
 _KEY_HEADER = "__temporal_stream_key"
@@ -80,6 +83,7 @@ def __init__(
         )
         self._temporal_stream_backend = StreamBackend(self._extension)
         self._converter = temporalio.converter.DataConverter.default
+        self._read_activity = _read_activity(self)
         super().__init__(name, interceptors=[_StreamInterceptor(self)])
 
     @property
@@ -110,6 +114,8 @@ def configure_worker(
         client = config.get("client")
         if client is not None:
             self._converter = client.data_converter
+        # Workflow reads run as this Local Activity, so the Worker must have it.
+        config["activities"] = [*config.get("activities", []), self._read_activity]
         runner = config.get("workflow_runner")
         if isinstance(runner, SandboxedWorkflowRunner):
             # The interceptor and the facade must share the run's Stream state, also when the
@@ -183,6 +189,55 @@ async def on_workflow_activation_complete(
         self._temporal_stream_backend.call("BufferWorkflowCalls", request)
 
 
+def read_activity_name(backend: str) -> str:
+    """The name of the Local Activity that reads a page of a backend for Workflow code."""
+    return f"__temporal_stream_read:{backend}"
+
+
+def _read_activity(plugin: StreamPlugin) -> Callable[..., Any]:
+    @temporalio.activity.defn(name=read_activity_name(plugin.backend_name))
+    async def read_page(
+        request: stream_pb.ReadRecordsRequest,
+    ) -> stream_pb.ReadRecordsResponse:
+        request.namespace = temporalio.activity.info().namespace
+        try:
+            page = await plugin._temporal_stream_backend.call_async(
+                "ReadRecords", request, stream_pb.ReadRecordsResponse
+            )
+        except _errors.StreamError as err:
+            # The reason goes in the details, so Workflow code gets the same Stream error.
+            raise temporalio.exceptions.ApplicationError(
+                str(err.message),
+                err.reason,
+                type=type(err).__name__,
+                non_retryable=err.non_retryable,
+            ) from None
+        # Workflow code can't run the codec, so the page's bodies go to it decoded. The page
+        # as a whole goes through the codec as the Local Activity's result.
+        await _decode_page(plugin._converter, page)
+        return page
+
+    return read_page
+
+
+async def _decode_page(
+    converter: temporalio.converter.DataConverter, page: stream_pb.ReadRecordsResponse
+) -> None:
+    codec = converter.payload_codec
+    if codec is None:
+        return
+    payloads = [
+        payload
+        for record in page.records
+        for payload in (record.body, record.close_result)
+        if payload.ByteSize()
+    ]
+    if not payloads:
+        return
+    for payload, decoded in zip(payloads, await codec.decode(payloads)):
+        payload.CopyFrom(decoded)
+
+
 class RedisStreamPlugin(StreamPlugin):
     """Carries Stream calls to Redis, through the temporalio-streams runtime in the bridge.
 
@@ -579,6 +634,9 @@ def _take_close_with_run(
 
 class _WorkflowOutbound(temporalio.worker.WorkflowOutboundInterceptor):
     _temporal_stream_transport = True
+    # The runtime can't park a read yet, so a reader pauses on a timer after an empty page.
+    _temporal_stream_parks_reads = False
+    _temporal_stream_read_wait = _READ_WAIT
 
     def __init__(
         self,
@@ -629,6 +687,25 @@ async def start_system_nexus_operation(
         )
         return _DoneHandle(result)
 
+    async def _temporal_stream_read_page(
+        self, request: stream_pb.ReadRecordsRequest
+    ) -> stream_pb.ReadRecordsResponse:
+        # The page is the Local Activity's result, so it is in History.
+        try:
+            return await temporalio.workflow.execute_local_activity(
+                read_activity_name(self._temporal_stream_backend_name),
+                request,
+                result_type=stream_pb.ReadRecordsResponse,
+                start_to_close_timeout=_READ_WAIT + _READ_MARGIN,
+            )
+        except (
+            temporalio.exceptions.ActivityError,
+            temporalio.exceptions.ApplicationError,
+        ) as err:
+            # A Local Activity's failure may come as its application error, with no
+            # ActivityError around it.
+            raise _workflow_reader.error_of(err) from None
+
     def continue_as_new(self, input: temporalio.worker.ContinueAsNewInput) -> NoReturn:
         open_streams = {
             stream_id: run_stream.backend
diff --git a/temporalio/contrib/streams/workflow.py b/temporalio/contrib/streams/workflow.py
index 859ce5859..6fd109b1d 100644
--- a/temporalio/contrib/streams/workflow.py
+++ b/temporalio/contrib/streams/workflow.py
@@ -13,17 +13,20 @@
 import asyncio
 import contextvars
 import weakref
-from collections.abc import Coroutine
+from collections.abc import AsyncIterator, Coroutine, Sequence
 from dataclasses import dataclass, field
 from datetime import timedelta
 from types import TracebackType
 from typing import Any, Generic, TypeVar, overload
 
+import temporalio.api.streamservice.v1 as stream_pb
 import temporalio.exceptions
 import temporalio.workflow
-from temporalio.contrib.streams import _errors, _ids
+from temporalio.contrib.streams import _errors, _ids, _workflow_reader
+from temporalio.contrib.streams._access import read_spec
 from temporalio.contrib.streams._defn import TopicDefinition, definition_of
 from temporalio.contrib.streams._handle import StreamHandle
+from temporalio.contrib.streams._workflow_reader import Item
 from temporalio.contrib.streams.system import create_stream as _create_stream
 from temporalio.contrib.streams.system import models
 from temporalio.contrib.streams.system._resources.stream import Stream as _Stream
@@ -192,9 +195,24 @@ def publish_nowait(self, *values: T) -> None:
         """
         self._stream._append(self._name, values)
 
+    def subscribe(
+        self,
+        *,
+        from_offset: int | None = None,
+        wait_new_records: bool = True,
+        create_wait: timedelta | None = None,
+    ) -> AsyncIterator[Item]:
+        """Read this Topic only. See :py:meth:`WorkflowStream.subscribe`."""
+        return self._stream.subscribe(
+            self,
+            from_offset=from_offset,
+            wait_new_records=wait_new_records,
+            create_wait=create_wait,
+        )
+
 
 class WorkflowStream(StreamHandle[E]):
-    """A Stream in Workflow code. Workflow code writes and closes it, and can't read it.
+    """A Stream in Workflow code. Workflow code writes, reads and closes it.
 
     Pass it to an Activity, a child Workflow or the next run of continue-as-new as an
     argument. Use it in ``async with`` to close it at the end of the block.
@@ -242,6 +260,59 @@ def topic(
         name, topic_type = self._topic_spec(topic, type)
         return WorkflowTopic(self, name, topic_type)
 
+    def subscribe(
+        self,
+        *topics: str | TopicDefinition[Any] | WorkflowTopic[Any],
+        from_offset: int | None = None,
+        wait_new_records: bool = True,
+        prefixes: Sequence[str] = (),
+        create_wait: timedelta | None = None,
+    ) -> AsyncIterator[Item]:
+        """Read the Stream's items in order, in Workflow code.
+
+        It is an async iterator, as on the client, and it yields the same ``Record``,
+        ``Superseded`` and ``Closed`` items. Each page is a call whose answer is in History,
+        so a replay reads the same items. On a native Stream a page is a System Nexus
+        ``ReadRecords`` operation, and a read at the end of the Stream waits on the server
+        until a Record comes or the Stream closes. On Redis a page is a Local Activity that
+        waits a short time for new Records, and a page with none is followed by a durable
+        timer from 1 s up to 60 s.
+
+        A read sees this run's writes once they went out: at the end of the Workflow Task,
+        or on Redis before the page's Local Activity.
+
+        Args:
+            topics: Read these Topics. With no Topics and no prefixes, read every Topic.
+            from_offset: Start at this offset, inclusive. Defaults to the first Record.
+            wait_new_records: Wait for new Records at the end of the Stream. When False,
+                the read ends at the newest Record.
+            prefixes: Also read the Topics whose names start with one of these.
+            create_wait: How long to wait for a Stream that nobody created yet. Defaults to
+                no wait.
+
+        Raises:
+            StreamNotFoundError: The Stream does not exist, or it was not created in time.
+            ReadOnlyContextError: The code must not change state, such as a query handler.
+            RuntimeError: No Stream plugin is on the Worker, or the Stream is of a backend
+                that the Worker does not have. It fails the Workflow Task, not the run.
+        """
+        names, types = read_spec(self, topics)
+        reader = _workflow_reader.reader_of(_check_backend(self.backend, "read"))
+        return _workflow_reader.read(
+            reader,
+            stream_pb.StreamRef(
+                id=self.id,
+                workflow_id=self.workflow_id or "",
+                first_execution_run_id=self.first_execution_run_id or "",
+            ),
+            topics=names,
+            prefixes=prefixes,
+            types=types,
+            from_offset=from_offset,
+            wait_new_records=wait_new_records,
+            create_wait=create_wait,
+        )
+
     def _check(self, action: str, *, backend: str | None = None) -> _RunStream:
         if temporalio.workflow.unsafe.is_read_only():
             raise _errors.ReadOnlyContextError(
diff --git a/tests/contrib/streams/conftest.py b/tests/contrib/streams/conftest.py
index 7a29e0b64..0be5d3ff1 100644
--- a/tests/contrib/streams/conftest.py
+++ b/tests/contrib/streams/conftest.py
@@ -1,3 +1,4 @@
+import os
 import shutil
 import socket
 import subprocess
@@ -8,6 +9,7 @@
 import pytest
 
 from temporalio.client import Client
+from temporalio.contrib.streams.native import NativeStreamPlugin
 from temporalio.contrib.streams.redis import RedisStreamPlugin, StreamPlugin
 from temporalio.contrib.streams.testing import MemoryStreamPlugin
 
@@ -63,3 +65,20 @@ async def stream_client(
     config = client.config()
     config["plugins"] = [plugin]
     yield Client(**config)
+
+
+# A server that serves the Stream service, such as a dev server of moetemp/temporal with the
+# native Stream backend. The native lane runs only when it is set.
+NATIVE_TARGET = os.environ.get("TEMPORAL_STREAMS_NATIVE_TARGET", "")
+NATIVE_NAMESPACE = os.environ.get("TEMPORAL_STREAMS_NATIVE_NAMESPACE", "default")
+
+
+@pytest.fixture
+async def native_client() -> Client:
+    if not NATIVE_TARGET:
+        pytest.skip(
+            "TEMPORAL_STREAMS_NATIVE_TARGET names no server with the Stream service"
+        )
+    return await Client.connect(
+        NATIVE_TARGET, namespace=NATIVE_NAMESPACE, plugins=[NativeStreamPlugin()]
+    )
diff --git a/tests/contrib/streams/fake_native.py b/tests/contrib/streams/fake_native.py
index 28592a10d..c0179413a 100644
--- a/tests/contrib/streams/fake_native.py
+++ b/tests/contrib/streams/fake_native.py
@@ -21,6 +21,7 @@
 import temporalio.api.streamservice.v1 as stream_pb
 import temporalio.client
 import temporalio.converter
+import temporalio.exceptions
 import temporalio.nexus.system
 import temporalio.worker
 import temporalio.workflow
@@ -326,8 +327,24 @@ async def start_system_nexus_operation(
         request = _REQUESTS[input.operation_name].FromString(
             converter.to_payload(input.input).data
         )
-        request.namespace = temporalio.workflow.info().namespace
-        response = self._service.handle(input.operation_name, request)
+        info = temporalio.workflow.info()
+        request.namespace = info.namespace
+        if input.operation_name == "AppendRecords":
+            # The server knows the caller, so it sets the Workflow's Producer.
+            request.producer.CopyFrom(
+                stream_pb.Producer(
+                    kind=stream_pb.ProducerKind.PRODUCER_KIND_WORKFLOW,
+                    id=f"workflow:{info.run_id}",
+                    run_id=info.run_id,
+                )
+            )
+        response = self._answer(input.operation_name, request)
+        if input.operation_name == "ReadRecords" and request.HasField("wait"):
+            # The server parks the read. Here a durable timer polls the store instead.
+            until = temporalio.workflow.now() + request.wait.ToTimedelta()
+            while not response.records and temporalio.workflow.now() < until:
+                await temporalio.workflow.sleep(0.2)
+                response = self._answer(input.operation_name, request)
         payload = temporalio.api.common.v1.Payload(
             metadata={
                 "encoding": b"binary/protobuf",
@@ -340,6 +357,25 @@ async def start_system_nexus_operation(
         await asyncio.sleep(0)
         return _Answered(converter.from_payload(payload, input.output_type or Any))
 
+    def _answer(self, operation: str, request: Any) -> Any:
+        try:
+            return self._service.handle(operation, request)
+        except FakeError as err:
+            # A failed operation's cause is the handler's application error, with the Stream
+            # error's class name as its type and the reason as its detail.
+            failed = temporalio.exceptions.NexusOperationError(
+                err.reason,
+                scheduled_event_id=0,
+                endpoint="__temporal_system",
+                service=SERVICE,
+                operation=operation,
+                operation_token="",
+            )
+            name = "".join(part.title() for part in err.reason.split("_")) + "Error"
+            raise failed from temporalio.exceptions.ApplicationError(
+                err.reason, err.reason, type=name, non_retryable=True
+            )
+
 
 class _Answered(temporalio.workflow.NexusOperationHandle[Any]):
     def __init__(self, result: Any) -> None:
diff --git a/tests/contrib/streams/test_facade.py b/tests/contrib/streams/test_facade.py
index b3c178e1c..b3406d0b8 100644
--- a/tests/contrib/streams/test_facade.py
+++ b/tests/contrib/streams/test_facade.py
@@ -162,8 +162,10 @@ async def test_each_client_handle_is_its_own_producer(client: Client):
     assert len(producers) == 2
 
 
-def test_workflow_stream_has_no_reads():
-    assert not hasattr(streams.WorkflowStream, "subscribe")
+def test_workflow_stream_reads_only_by_subscribe():
+    # A read's pages are in History. An offset that a replay could not give again is not.
+    assert hasattr(streams.WorkflowStream, "subscribe")
+    assert hasattr(streams.WorkflowTopic, "subscribe")
     assert not hasattr(streams.WorkflowStream, "get_offset")
 
 
diff --git a/tests/contrib/streams/test_native_server.py b/tests/contrib/streams/test_native_server.py
new file mode 100644
index 000000000..dcad9eb26
--- /dev/null
+++ b/tests/contrib/streams/test_native_server.py
@@ -0,0 +1,101 @@
+"""The native lane: the same calls against a server that serves the Stream service.
+
+Set ``TEMPORAL_STREAMS_NATIVE_TARGET`` to its address to run it.
+"""
+
+import concurrent.futures
+import uuid
+from datetime import timedelta
+
+from temporalio.client import Client
+from temporalio.contrib import streams
+from temporalio.contrib.streams import workflow as stream_workflow
+from temporalio.contrib.streams.native import NativeStreamPlugin
+from temporalio.contrib.streams.redis import RedisStreamPlugin
+from temporalio.worker import Replayer
+from tests.contrib.streams.test_native import (
+    Events,
+    WritesNativeAndRedis,
+    write_native,
+    write_native_sync,
+)
+from tests.contrib.streams.test_workflow_reads import (
+    Reader,
+    ReadInput,
+    Writer,
+)
+from tests.helpers import new_worker
+
+
+async def test_native_client_round_trip(native_client: Client):
+    stream = await streams.StreamClient(native_client).create(
+        Events, id=f"native-{uuid.uuid4()}", auto_close=timedelta(minutes=5)
+    )
+    assert await stream.topic(Events.tokens).publish("a", "b")
+    await stream.close(result="done")
+    items = [item async for item in stream.subscribe()]
+    assert [item.data for item in items if isinstance(item, streams.Record)] == [
+        "a",
+        "b",
+    ]
+    assert isinstance(items[-1], streams.Closed) and items[-1].result == "done"
+
+
+async def test_native_workflow_reads_and_replays(native_client: Client):
+    async with new_worker(native_client, Writer, Reader) as worker:
+        writer = await native_client.start_workflow(
+            Writer.run, None, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue
+        )
+        stream = streams.StreamClient(native_client).workflow_stream(writer, Events)
+        reader = await native_client.start_workflow(
+            Reader.run,
+            ReadInput(stream_workflow.WorkflowStream.from_handle(stream)),
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert await reader.result() == [
+            "turn user hi",
+            "a",
+            "b",
+            "c",
+            "closed written",
+        ]
+    replayer = Replayer(workflows=[Writer, Reader], plugins=[NativeStreamPlugin()])
+    await replayer.replay_workflow(await writer.fetch_history())
+    await replayer.replay_workflow(await reader.fetch_history())
+
+
+async def test_redis_and_native_on_a_native_server(
+    native_client: Client, redis_url: str
+):
+    config = native_client.config()
+    config["plugins"] = [
+        RedisStreamPlugin(redis_url, key_prefix=f"t-{uuid.uuid4()}"),
+        NativeStreamPlugin(),
+    ]
+    two = Client(**config)
+    async with new_worker(
+        two,
+        WritesNativeAndRedis,
+        activities=[write_native, write_native_sync],
+        activity_executor=concurrent.futures.ThreadPoolExecutor(max_workers=2),
+    ) as worker:
+        handle = await two.start_workflow(
+            WritesNativeAndRedis.run,
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert await handle.result() == ["redis", "native"]
+    new = streams.StreamClient(two).workflow_stream(
+        handle, Events, name="new", backend="native"
+    )
+    items = [item async for item in new.subscribe()]
+    assert [item.data for item in items if isinstance(item, streams.Record)] == [
+        "workflow 1",
+        "workflow 2",
+        "activity",
+        "sync activity",
+        "last",
+    ]
+    assert isinstance(items[-1], streams.Closed)
+    assert items[-1].result == ["redis", "native"]
diff --git a/tests/contrib/streams/test_workflow_reads.py b/tests/contrib/streams/test_workflow_reads.py
new file mode 100644
index 000000000..cb805333b
--- /dev/null
+++ b/tests/contrib/streams/test_workflow_reads.py
@@ -0,0 +1,275 @@
+import dataclasses
+import uuid
+from collections.abc import Coroutine
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Any, cast
+
+import pytest
+
+from temporalio import workflow
+from temporalio.client import Client, WorkflowFailureError
+from temporalio.contrib import streams
+from temporalio.contrib.streams import workflow as stream_workflow
+from temporalio.contrib.streams.native import NativeStreamPlugin
+from temporalio.contrib.streams.redis import StreamPlugin
+from temporalio.converter import DataConverter
+from temporalio.exceptions import ApplicationError
+from temporalio.worker import Replayer
+from tests.contrib.streams.fake_native import (
+    FakeStreamService,
+    FakeSystemNexus,
+)
+from tests.contrib.streams.test_redis import _Rot13Codec
+from tests.helpers import new_worker
+
+
+@dataclass
+class Turn:
+    kind: str
+    text: str = ""
+
+
+@streams.defn(name="events")
+class Events:
+    turns = streams.topic(Turn)
+    tokens = streams.topic(str)
+
+
+@workflow.defn
+class Writer:
+    @workflow.run
+    async def run(self, backend: str | None) -> str:
+        stream = await stream_workflow.create_stream(Events, backend=backend)
+        await stream.topic(Events.turns).publish(Turn("user", "hi"))
+        for token in ("a", "b", "c"):
+            # Each token goes out in its own Workflow Task, so the reader waits for some.
+            await workflow.sleep(0.3)
+            await stream.topic(Events.tokens).publish(token)
+        return "written"
+
+
+@dataclass
+class ReadInput:
+    stream: stream_workflow.WorkflowStream[Events]
+    tokens_only: bool = False
+    wait_new_records: bool = True
+
+
+@workflow.defn
+class Reader:
+    @workflow.run
+    async def run(self, input: ReadInput) -> list[Any]:
+        seen: list[Any] = []
+        items = (
+            input.stream.topic(Events.tokens).subscribe(
+                wait_new_records=input.wait_new_records,
+                create_wait=timedelta(seconds=20),
+            )
+            if input.tokens_only
+            else input.stream.subscribe(
+                wait_new_records=input.wait_new_records,
+                create_wait=timedelta(seconds=20),
+            )
+        )
+        async for item in items:
+            # The Topic's type decodes each body in Workflow code.
+            if isinstance(item, streams.Record) and isinstance(item.data, Turn):
+                seen.append(f"turn {item.data.kind} {item.data.text}")
+            elif isinstance(item, streams.Record):
+                seen.append(item.data)
+            elif isinstance(item, streams.Closed):
+                seen.append(f"closed {item.result}")
+        return seen
+
+
+def plugins_of(client: Client) -> list[Any]:
+    return [
+        plugin
+        for plugin in client.config()["plugins"]
+        if isinstance(plugin, (StreamPlugin, NativeStreamPlugin))
+    ]
+
+
+async def write_and_read(
+    client: Client, backend: str | None, **read_options: Any
+) -> tuple[list[Any], Any]:
+    async with new_worker(client, Writer, Reader) as worker:
+        writer = await client.start_workflow(
+            Writer.run,
+            backend,
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        stream = streams.StreamClient(client).workflow_stream(
+            writer, Events, backend=backend
+        )
+        reader = await client.start_workflow(
+            Reader.run,
+            ReadInput(
+                stream_workflow.WorkflowStream.from_handle(stream), **read_options
+            ),
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        result = await reader.result()
+        assert await writer.result() == "written"
+    return result, reader
+
+
+async def test_workflow_reads_another_workflows_stream(stream_client: Client):
+    seen, reader = await write_and_read(stream_client, None)
+    assert seen == ["turn user hi", "a", "b", "c", "closed written"]
+    # Each page is in History, so a replay reads the same items.
+    replayer = Replayer(workflows=[Reader], plugins=plugins_of(stream_client))
+    await replayer.replay_workflow(await reader.fetch_history())
+
+
+async def test_workflow_reads_one_topic(stream_client: Client):
+    seen, _ = await write_and_read(stream_client, None, tokens_only=True)
+    assert seen == ["a", "b", "c", "closed written"]
+
+
+@workflow.defn
+class ReadsMissing:
+    @workflow.run
+    async def run(self, stream_id: str) -> str:
+        stream = stream_workflow.WorkflowStream[Events](id=stream_id)
+        try:
+            async for _ in stream.subscribe():
+                pass
+        except streams.StreamNotFoundError as err:
+            return f"not found: {err.reason}"
+        return "read"
+
+
+async def test_a_missing_stream_raises_in_workflow_code(stream_client: Client):
+    async with new_worker(stream_client, ReadsMissing) as worker:
+        result = await stream_client.execute_workflow(
+            ReadsMissing.run,
+            f"missing-{uuid.uuid4()}",
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+    assert result == "not found: stream_not_found"
+
+
+@workflow.defn
+class ReadsOwnWrites:
+    @workflow.run
+    async def run(self) -> list[Any]:
+        stream = await stream_workflow.create_stream(Events, close_with_run=False)
+        await stream.topic(Events.tokens).publish("mine")
+        # A read's Local Activity first writes the run's buffer, so the read sees it.
+        return [
+            item.data
+            async for item in stream.subscribe(wait_new_records=False)
+            if isinstance(item, streams.Record)
+        ]
+
+
+async def test_a_redis_read_sees_the_runs_own_writes(stream_client: Client):
+    async with new_worker(stream_client, ReadsOwnWrites) as worker:
+        assert await stream_client.execute_workflow(
+            ReadsOwnWrites.run,
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        ) == ["mine"]
+
+
+@workflow.defn
+class ReadsInQuery:
+    def __init__(self) -> None:
+        self._stream: stream_workflow.WorkflowStream[Events] | None = None
+
+    @workflow.run
+    async def run(self) -> None:
+        self._stream = await stream_workflow.create_stream(Events)
+        await workflow.wait_condition(lambda: False)
+
+    @workflow.query
+    def read(self) -> str:
+        assert self._stream is not None
+        items = self._stream.subscribe()
+        try:
+            # A query can't await, so it takes the read's first step by hand.
+            cast(Coroutine[Any, Any, Any], items.__anext__()).send(None)
+        except streams.ReadOnlyContextError as err:
+            return str(err)
+        except StopIteration:
+            pass
+        return "read"
+
+
+async def test_a_read_in_a_query_raises(stream_client: Client):
+    async with new_worker(stream_client, ReadsInQuery) as worker:
+        handle = await stream_client.start_workflow(
+            ReadsInQuery.run, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue
+        )
+        assert "read-only" in await handle.query(ReadsInQuery.read)
+        await handle.terminate()
+
+
+async def test_native_workflow_reads(client: Client):
+    # Until a dev server serves the Stream service, the System Nexus stand-in answers the
+    # Workflow's calls and parks a read on a timer.
+    service = FakeStreamService()
+    config = client.config()
+    config["interceptors"] = [FakeSystemNexus(service)]
+    config["plugins"] = [NativeStreamPlugin()]
+    native_client = Client(**config)
+    async with new_worker(native_client, Writer, Reader) as worker:
+        writer = await native_client.start_workflow(
+            Writer.run, None, id=f"wf-{uuid.uuid4()}", task_queue=worker.task_queue
+        )
+        assert writer.first_execution_run_id
+        stream = stream_workflow.WorkflowStream[Events](
+            id=f"{writer.id}:{writer.first_execution_run_id}:events",
+            workflow_id=writer.id,
+            first_execution_run_id=writer.first_execution_run_id,
+            backend="native",
+        )
+        seen = await native_client.execute_workflow(
+            Reader.run,
+            ReadInput(stream),
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+    assert seen == ["turn user hi", "a", "b", "c", "closed written"]
+    reads = [request for name, request in service.calls if name == "ReadRecords"]
+    assert all(request.wait.ToTimedelta() == timedelta(seconds=60) for request in reads)
+
+
+@workflow.defn
+class FailsOnStreamError:
+    @workflow.run
+    async def run(self, stream_id: str) -> None:
+        stream = stream_workflow.WorkflowStream[Events](id=stream_id)
+        async for _ in stream.subscribe():
+            pass
+
+
+async def test_an_uncaught_read_error_fails_the_run(stream_client: Client):
+    # A Stream error is in History, so it fails the run, not the Workflow Task.
+    async with new_worker(stream_client, FailsOnStreamError) as worker:
+        with pytest.raises(WorkflowFailureError) as failure:
+            await stream_client.execute_workflow(
+                FailsOnStreamError.run,
+                f"missing-{uuid.uuid4()}",
+                id=f"wf-{uuid.uuid4()}",
+                task_queue=worker.task_queue,
+            )
+    assert isinstance(failure.value.cause, ApplicationError)
+    assert failure.value.cause.type == "StreamNotFoundError"
+
+
+async def test_workflow_reads_go_through_the_codec(stream_client: Client):
+    # The Store keeps encoded bodies, and Workflow code can't run the codec, so the read's
+    # Local Activity decodes them.
+    config = stream_client.config()
+    config["data_converter"] = dataclasses.replace(
+        DataConverter.default, payload_codec=_Rot13Codec()
+    )
+    coded = Client(**config)
+    seen, _ = await write_and_read(coded, None)
+    assert seen == ["turn user hi", "a", "b", "c", "closed written"]

From 28f648900e1eced33f94e6b4a3edd84e0a69fbd2 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 01:55:57 -0700
Subject: [PATCH 2/6] Mapped a Workflow read failure by a reason in its type or
 message.

---
 temporalio/contrib/streams/_errors.py        | 11 +++++++++--
 tests/contrib/streams/test_workflow_reads.py | 13 +++++++++++++
 2 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/temporalio/contrib/streams/_errors.py b/temporalio/contrib/streams/_errors.py
index 82a6fce84..a84ad559e 100644
--- a/temporalio/contrib/streams/_errors.py
+++ b/temporalio/contrib/streams/_errors.py
@@ -125,12 +125,19 @@ def from_application_error(
 ) -> StreamError | None:
     """The Stream error that an application error of a failed call carries, if any.
 
-    A Stream error's type is its class name, and its first detail is its reason.
+    A Stream error's type is its class name, and its first detail is its reason. A server may
+    give the reason as the type, or at the start of the message, such as ``stream_closed: ...``.
     """
+    token = err.message.split(":", 1)[0].strip()
+    reason = err.details[0] if err.details and isinstance(err.details[0], str) else ""
     cls = _BY_NAME.get(err.type or "")
+    if cls is None:
+        reason = next(
+            (name for name in (err.type or "", token) if name in _BY_REASON), ""
+        )
+        cls = _BY_REASON.get(reason)
     if cls is None:
         return None
-    reason = err.details[0] if err.details and isinstance(err.details[0], str) else ""
     return cls(err.message, reason=reason)
 
 
diff --git a/tests/contrib/streams/test_workflow_reads.py b/tests/contrib/streams/test_workflow_reads.py
index cb805333b..3d568fec5 100644
--- a/tests/contrib/streams/test_workflow_reads.py
+++ b/tests/contrib/streams/test_workflow_reads.py
@@ -10,6 +10,7 @@
 from temporalio import workflow
 from temporalio.client import Client, WorkflowFailureError
 from temporalio.contrib import streams
+from temporalio.contrib.streams import _errors
 from temporalio.contrib.streams import workflow as stream_workflow
 from temporalio.contrib.streams.native import NativeStreamPlugin
 from temporalio.contrib.streams.redis import StreamPlugin
@@ -273,3 +274,15 @@ async def test_workflow_reads_go_through_the_codec(stream_client: Client):
     coded = Client(**config)
     seen, _ = await write_and_read(coded, None)
     assert seen == ["turn user hi", "a", "b", "c", "closed written"]
+
+
+def test_a_server_failure_maps_by_its_reason():
+    by_type = _errors.from_application_error(
+        ApplicationError("closed", type="stream_closed")
+    )
+    assert isinstance(by_type, streams.StreamClosedError)
+    by_message = _errors.from_application_error(
+        ApplicationError("stream_not_found: Nobody created the Stream")
+    )
+    assert isinstance(by_message, streams.StreamNotFoundError)
+    assert _errors.from_application_error(ApplicationError("other")) is None

From 2eaa5eb3cb46c689a18dd7f706d667358fb93663 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 02:11:02 -0700
Subject: [PATCH 3/6] Resolved the import lines that the last merge left
 marked.

---
 temporalio/contrib/streams/native.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/temporalio/contrib/streams/native.py b/temporalio/contrib/streams/native.py
index 118eaa691..a6c64eb5d 100644
--- a/temporalio/contrib/streams/native.py
+++ b/temporalio/contrib/streams/native.py
@@ -10,12 +10,8 @@
 import dataclasses
 import itertools
 import weakref
-<<<<<<< HEAD
-from collections.abc import Coroutine, Generator
-from datetime import timedelta
-=======
 from collections.abc import Callable, Coroutine, Generator
->>>>>>> moe/jetty/v2/j2-native-plugin
+from datetime import timedelta
 from typing import Any, NoReturn, TypeVar
 
 import google.protobuf.message

From cd33cbb9433af6142ff16be7c3eff8d293d73bca Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 02:40:30 -0700
Subject: [PATCH 4/6] Kept a native read line within 100 columns.

---
 temporalio/contrib/streams/native.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/temporalio/contrib/streams/native.py b/temporalio/contrib/streams/native.py
index f23310aeb..d2815516c 100644
--- a/temporalio/contrib/streams/native.py
+++ b/temporalio/contrib/streams/native.py
@@ -27,6 +27,8 @@
 
 NATIVE_EXTENSION_NAME = "temporalio-streams-native"
 
+_WAIT_COMPLETED = temporalio.workflow.NexusOperationCancellationType.WAIT_COMPLETED
+
 
 class _NativeBackend(_ExtensionStreamBackend):
     def for_client(self, client: temporalio.client.Client) -> _NativeBackend:
@@ -59,7 +61,7 @@ async def _temporal_stream_read_page(
                     schedule_to_close_timeout=None,
                     schedule_to_start_timeout=None,
                     start_to_close_timeout=None,
-                    cancellation_type=temporalio.workflow.NexusOperationCancellationType.WAIT_COMPLETED,
+                    cancellation_type=_WAIT_COMPLETED,
                     headers=None,
                     summary=None,
                 )

From 924d14bda19913fd744435d5c772d132c8a5ce41 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 11:06:35 -0700
Subject: [PATCH 5/6] Set the query test's Stream before a query can come.

---
 tests/contrib/streams/test_workflow_reads.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/contrib/streams/test_workflow_reads.py b/tests/contrib/streams/test_workflow_reads.py
index dcb5e97f9..f5cf4d207 100644
--- a/tests/contrib/streams/test_workflow_reads.py
+++ b/tests/contrib/streams/test_workflow_reads.py
@@ -179,7 +179,8 @@ def __init__(self) -> None:
 
     @workflow.run
     async def run(self) -> None:
-        self._stream = await stream_workflow.create_stream(Events)
+        # Set in the first task, since a query may come before a create's Local Activity ends.
+        self._stream = stream_workflow.create_stream_nowait(Events)
         await workflow.wait_condition(lambda: False)
 
     @workflow.query

From cc79c70d11e19cdab80c5aa53653dbe1a4d9a5b3 Mon Sep 17 00:00:00 2001
From: Mohammad Dashti <mohammad.dashti@temporal.io>
Date: Thu, 22 Oct 2026 14:22:08 -0700
Subject: [PATCH 6/6] Named the native test Streams with `streams.with_name`.

---
 tests/contrib/streams/test_native_server.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/tests/contrib/streams/test_native_server.py b/tests/contrib/streams/test_native_server.py
index 745507ff6..be524ff17 100644
--- a/tests/contrib/streams/test_native_server.py
+++ b/tests/contrib/streams/test_native_server.py
@@ -37,13 +37,13 @@ def write_native_sync(stream: streams.SyncActivityStream[Events]) -> None:
 class WritesNativeAndRedis:
     @workflow.run
     async def run(self) -> list[str | None]:
-        on_redis = await stream_workflow.create_stream(Events, name="old")
+        on_redis = await stream_workflow.create_stream(streams.with_name(Events, "old"))
         on_native = await stream_workflow.create_stream(
-            Events, name="new", backend="native"
+            streams.with_name(Events, "new"), backend="native"
         )
         await on_redis.topic(Events.tokens).publish("redis")
         await on_native.topic(Events.tokens).publish("workflow 1", "workflow 2")
-        # The Activities write after the run's buffer went out, as on Redis.
+        # The Activities write after the run's own writes, as on Redis.
         await workflow.execute_activity(
             write_native,
             streams.ActivityStream.from_handle(on_native),
@@ -118,7 +118,7 @@ async def test_redis_and_native_on_a_native_server(
         )
         assert await handle.result() == ["redis", "native"]
     new = streams.StreamClient(two).workflow_stream(
-        handle, Events, name="new", backend="native"
+        handle, streams.with_name(Events, "new"), backend="native"
     )
     items = [item async for item in new.subscribe()]
     assert [item.data for item in items if isinstance(item, streams.Record)] == [