From cae0e2a1c50149f2bb405355f8601a4aa818d373 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:21:07 -0700 Subject: [PATCH 1/5] Added the Activity and client stream handles over Core. --- changelog/added/pirouetting-platypus.md | 1 + temporalio/contrib/streams/__init__.py | 23 +- temporalio/contrib/streams/_handles.py | 569 ++++++++++++++++++ tests/contrib/streams/test_outside_handles.py | 450 ++++++++++++++ 4 files changed, 1042 insertions(+), 1 deletion(-) create mode 100644 changelog/added/pirouetting-platypus.md create mode 100644 temporalio/contrib/streams/_handles.py create mode 100644 tests/contrib/streams/test_outside_handles.py diff --git a/changelog/added/pirouetting-platypus.md b/changelog/added/pirouetting-platypus.md new file mode 100644 index 0000000000..7171967de3 --- /dev/null +++ b/changelog/added/pirouetting-platypus.md @@ -0,0 +1 @@ +Add experimental `temporalio.contrib.streams` so Activities and clients can append ordered records to a Workflow's stream in Redis, and outside readers can follow it from a cursor. diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index d672e8a52e..c6976630bb 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -32,7 +32,16 @@ A store is configured as a plugin: ``Client.connect(..., plugins=[RedisStreams(...)])`` registers it on the -client and on every Worker built from that client. +client and on every Worker built from that client. Each +context then reaches a stream the same way: + +- An Activity reaches the stream of the Workflow that scheduled it with + :func:`activity_handle`, and writes as itself: its Activity id with the + scheduling run id, and its Temporal attempt. The handle is pinned to that + run, so a read on it ends when the run closes, Continue-as-New included. +- Any process holding a client reaches a Workflow's stream with + :func:`get_stream_handle`, and writes with a producer id and an attempt of + its own. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` @@ -53,6 +62,13 @@ StreamStorageError, StreamUnsupportedError, ) +from temporalio.contrib.streams._handles import ( + ActivityStreamHandle, + StreamHandle, + StreamProducer, + activity_handle, + get_stream_handle, +) from temporalio.contrib.streams._plugin import StreamStorePlugin from temporalio.contrib.streams._record import ( Cursor, @@ -64,6 +80,7 @@ from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic __all__ = [ + "ActivityStreamHandle", "BEGINNING", "Cursor", "DEFAULT_TOPIC", @@ -73,9 +90,11 @@ "StreamCursorError", "StreamError", "StreamExpiredError", + "StreamHandle", "StreamNotFoundError", "StreamOutcomeUnknownError", "StreamOwnerKind", + "StreamProducer", "StreamProducerError", "StreamRecord", "StreamRecordError", @@ -86,5 +105,7 @@ "StreamTopic", "StreamUnsupportedError", "Supersession", + "activity_handle", + "get_stream_handle", "topic", ] diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py new file mode 100644 index 0000000000..e922484e30 --- /dev/null +++ b/temporalio/contrib/streams/_handles.py @@ -0,0 +1,569 @@ +"""Reaching a stream from an Activity or from client code. + +Every call goes to Core's stream service, which checks the owner, mints and +checks cursors, and synthesizes ``SUPERSEDED``. What stays here is what +needs the plaintext or the user's types: converting values, the two hashes +taken before the codec, and the codec itself. + +An Activity reaches the stream of the Workflow that scheduled it, and writes +as itself: Core derives its producer id as +``@`` and its Temporal attempt is the +producer attempt, so a retry is reported to readers as ``SUPERSEDED``. The +run is part of the id because a stream outlives a run, and Activity ids +repeat across the runs of a chain. + +Any process holding a client reaches a Workflow's stream by Workflow id and +writes with a producer id and attempt of its own. +""" + +from __future__ import annotations + +import asyncio +import weakref +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any, Generic, TypeVar, cast, overload + +import temporalio.activity +from temporalio.bridge.proto.streams import ( + ActivityProducer, + AppendRecord, + AppendRequest, + LatestRequest, + NamedProducer, + ReadRequest, + StreamAddress, + StreamOwnerKind, +) +from temporalio.client import Client +from temporalio.contrib.streams._body import batch_digest, content_hash, encode_bodies +from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._errors import StreamUnsupportedError +from temporalio.contrib.streams._plugin import ( + StreamStorePlugin, + call, + store_for_client, +) +from temporalio.contrib.streams._record import Cursor, RecordKind, StreamRecord +from temporalio.contrib.streams._ref import StreamRef +from temporalio.contrib.streams._topic import StreamTopic, resolve_topic +from temporalio.contrib.streams._wire import WireRecord, from_read, to_wire +from temporalio.converter import DataConverter, WorkflowSerializationContext + +__all__ = [ + "ActivityStreamHandle", + "StreamHandle", + "StreamProducer", + "activity_handle", + "get_stream_handle", +] + +T = TypeVar("T") + +_READ_WAIT = timedelta(seconds=30) + + +class StreamProducer(Generic[T]): + """Appends to one topic as one producer attempt. + + Every append is visible as soon as the store accepts it. Each record + carries the producer id, the attempt and a sequence that starts at one + for each attempt, which lets the store deduplicate a retry and lets a + reader tell a retry from a new attempt. + """ + + def __init__( + self, + handle: StreamHandle, + topic: str, + producer_id: str, + attempt: int, + activity: ActivityProducer | None = None, + ) -> None: + """Prefer :meth:`StreamHandle.producer`.""" + self._handle = handle + self._topic = topic + self._producer_id = producer_id + self._attempt = attempt + self._activity = activity + self._sequence = 1 + self._last = BEGINNING + # A batch reads the sequence, then awaits the codec and Core, then + # moves it on; calls one at a time keep each batch's sequences. + self._lock = asyncio.Lock() + + @property + def producer_id(self) -> str: + """Who this producer writes as.""" + return self._producer_id + + @property + def attempt(self) -> int: + """The attempt this producer writes.""" + return self._attempt + + async def append(self, *values: T) -> Cursor: + """Append ``values`` as one batch and return the cursor of its last record. + + The batch lands whole and in order. The producer moves its sequence + past the batch only when the append succeeds, so a call that raised + :class:`temporalio.contrib.streams.StreamOutcomeUnknownError` can be + repeated with the same values: the store answers a repeat of the + batch it holds last for this producer attempt with the original + position and writes nothing. + + An empty call writes nothing and returns the cursor of this + producer's last record, or ``BEGINNING`` when it has written none. + + Calls on one producer run one at a time, in the order they were + made, so concurrent calls take consecutive sequences. A call that was + cancelled may or may not have written its batch, and the producer + cannot tell: start a new attempt (a new producer with a higher + ``attempt``) rather than continuing this one. + + Raises: + StreamProducerError: The sequence was already used with different + content, or it is below the newest one the store holds for + this producer attempt. + StreamOutcomeUnknownError: The store may or may not have written + the batch. Retry the same values on this producer. + StreamRefusedError: The store refused the batch, for example + because it is out of memory. Nothing was written. + StreamClosedError: The stream refuses appends. + """ + async with self._lock: + if not values: + return self._last + return await self._write(RecordKind.DATA, values) + + async def finish(self) -> Cursor: + """Write ``FINISH`` for this producer on this topic and return its cursor. + + Says this producer has nothing more to send. It does not say the + work behind it succeeded, and it does not end anyone's read. + """ + async with self._lock: + return await self._write(RecordKind.FINISH, [None]) + + async def _write(self, kind: RecordKind, values: Any) -> Cursor: + converter = self._handle._converter + wires = [ + to_wire( + converter.payload_converter, + topic=self._topic, + kind=kind, + value=value, + producer_id=self._producer_id, + attempt=self._attempt, + sequence=self._sequence + index, + ) + for index, value in enumerate(values) + ] + request = AppendRequest( + stream=self._handle._address(self._topic), + sequence=self._sequence, + records=await _append_records(converter, wires), + digest=batch_digest(wires), + ) + if self._activity is not None: + request.activity.CopyFrom(self._activity) + else: + request.named.CopyFrom( + NamedProducer(producer_id=self._producer_id, attempt=self._attempt) + ) + service = await self._handle._plugin._service_for(self._handle._client) + response = await call(service.append(request)) + self._sequence += len(wires) + self._last = Cursor(response.last_cursor) + return self._last + + +async def _append_records( + converter: Any, wires: list[WireRecord] +) -> list[AppendRecord]: + bodies = iter( + await encode_bodies(converter, [w.body for w in wires if w.HasField("body")]) + ) + records = [] + for wire in wires: + record = AppendRecord(kind=wire.kind) + if wire.HasField("body"): + record.body.CopyFrom(next(bodies)) + record.content_hash = content_hash(wire.body) + records.append(record) + return records + + +class StreamHandle: + """One stream, addressed by topic, from outside Workflow code. + + A handle on a Workflow's stream follows the run chain unless its + :attr:`ref` is pinned to a run. A topic is a + :class:`temporalio.contrib.streams.StreamTopic`, which carries the record + type, or a plain string with ``result_type=`` for a name decided at + runtime. A call that names no topic addresses the topic of :attr:`ref`. + """ + + def __init__( + self, plugin: StreamStorePlugin, client: Client, ref: StreamRef + ) -> None: + """Prefer :func:`temporalio.contrib.streams.get_stream_handle`.""" + self._plugin = plugin + self._client = client + self._ref = ref + # The stream is the Workflow's, and the Workflow encodes its records with + # its own context, so every reader and producer on it uses that context. + self._converter: DataConverter = client.data_converter.with_context( + WorkflowSerializationContext( + namespace=client.namespace, workflow_id=ref.workflow_id + ) + ) + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, as data another process can open.""" + return self._ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """Yield the records on ``topic`` after ``after`` as they arrive. + + ``BEGINNING`` starts at the oldest record the topic retains, and + ``END`` at whatever is appended after the read starts. Any other + cursor came from a record of this stream, and the read resumes just + past it, so a reader that keeps the last cursor it handled and hands + it back sees every stored record once. A resumed read knows the + attempt of the record at its cursor, so a new attempt of that + producer still arrives as ``SUPERSEDED``. Earlier attempts of other + producers are not known to it. The read ends when the owner's run + chain is closed and every retained record has been delivered. The + result is a generator, so a caller that stops early calls + ``aclose()`` on it. + + The read starts when the generator first runs, not when this call + returns, and ``END`` is resolved then. A reader that must not miss + the next record, such as one about to send something the Workflow + answers on the stream, takes ``latest()`` first and reads after it:: + + start = await handle.latest() + await client.execute_update(...) + async for record in handle.read(after=start): + ... + + Raises: + ValueError: ``result_type`` was passed with a topic definition, + or the topic is empty. Raised by this call. + StreamCursorError: ``after`` came from another store or another + stream. Raised by the first iteration. + StreamExpiredError: ``after`` names a record the store no longer + retains, or the read fell behind retention while it ran. + StreamRecordError: A record did not decode. Read on from its + cursor to skip it. + StreamNotFoundError: The owner doesn't exist. + StreamStorageError: The store failed or couldn't be reached, as + a Redis blip during a long read does. + + Any of these ends the generator. Start a new read after the last + cursor you handled. It reads on from there, though it no longer knows + which earlier attempts of other producers it delivered. + """ + name, result_type = self._resolve(topic, result_type) + return self._read(name, after, result_type) + + async def _read( + self, topic: str, after: Cursor, result_type: type | None + ) -> AsyncGenerator[StreamRecord[Any], None]: + service = await self._plugin._service_for(self._client) + request = ReadRequest(stream=self._address(topic), after=after.token) + request.wait.FromTimedelta(_READ_WAIT) + while True: + response = await call(service.read(request)) + for record in response.records: + yield await from_read(self._converter, record, result_type) + if response.done: + return + request.after = response.cursor + request.state = response.state + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """The cursor of the newest record on ``topic``, or ``BEGINNING`` when empty. + + ``read(after=latest())`` yields only what is appended after this call + returned, which is how a client positions itself before it sends + something the Workflow answers on the stream. + """ + name, _ = self._resolve(topic, None) + service = await self._plugin._service_for(self._client) + response = await call(service.latest(LatestRequest(stream=self._address(name)))) + return Cursor(response.cursor) + + @overload + def producer( + self, *, topic: StreamTopic[T], producer_id: str, attempt: int + ) -> StreamProducer[T]: ... + + @overload + def producer( + self, *, topic: str | None = None, producer_id: str, attempt: int + ) -> StreamProducer[Any]: ... + + def producer( + self, + *, + topic: str | StreamTopic[Any] | None = None, + producer_id: str, + attempt: int, + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as ``producer_id`` in ``attempt``. + + A process that restarts its work raises ``attempt``, which starts a + new sequence and tells readers that what the earlier attempt wrote is + superseded. Keep one producer object per ``(topic, producer_id, + attempt)``. Each object numbers its own records from one. A second + object for the same session is refused as stale once the first moved + past its first batch. Before that, a second object's first batch with + the same content as the first object's is taken as a retry, and the + store writes nothing, with no error. + + Raises: + ValueError: ``producer_id`` is empty or ``attempt`` is below one. + """ + if not producer_id: + raise ValueError("producer_id must not be empty") + if isinstance(attempt, bool) or not isinstance(attempt, int) or attempt < 1: + raise ValueError(f"attempt must be an int of at least 1, got {attempt!r}") + name, _ = self._resolve(topic, None) + return StreamProducer(self, name, producer_id, attempt) + + def _resolve( + self, topic: str | StreamTopic[Any] | None, result_type: type | None + ) -> tuple[str, type | None]: + return resolve_topic(self._ref.topic if topic is None else topic, result_type) + + def _address(self, topic: str) -> StreamAddress: + return StreamAddress( + namespace=self._client.namespace, + owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, + workflow_id=self._ref.workflow_id, + run_id=self._ref.run_id or "", + topic=topic, + ) + + +class ActivityStreamHandle: + """The scheduling Workflow's stream, as the running Activity sees it. + + Reads and positions as :class:`temporalio.contrib.streams.StreamHandle` + does. :meth:`producer` writes as this Activity, which is the only + identity that lets readers tell this Activity's retry from a new + producer. + """ + + def __init__( + self, + inner: StreamHandle, + identity: ActivityProducer, + producers: dict[str, StreamProducer[Any]], + ) -> None: + """Prefer :func:`temporalio.contrib.streams.activity_handle`.""" + self._inner = inner + self._identity = identity + self._producers = producers + + @property + def ref(self) -> StreamRef: + """The stream this handle is on, pinned to the scheduling run.""" + return self._inner.ref + + @overload + def read( + self, *, topic: StreamTopic[T], after: Cursor = ... + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: type[T], + ) -> AsyncGenerator[StreamRecord[T], None]: ... + + @overload + def read( + self, + *, + topic: str | None = None, + after: Cursor = ..., + result_type: None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: ... + + def read( + self, + *, + topic: str | StreamTopic[Any] | None = None, + after: Cursor = BEGINNING, + result_type: type | None = None, + ) -> AsyncGenerator[StreamRecord[Any], None]: + """See :meth:`temporalio.contrib.streams.StreamHandle.read`.""" + # One call forwards every overload, which the checker can't match one by one. + inner = cast(Any, self._inner) + return inner.read(topic=topic, after=after, result_type=result_type) + + async def latest(self, *, topic: str | StreamTopic[Any] | None = None) -> Cursor: + """See :meth:`temporalio.contrib.streams.StreamHandle.latest`.""" + return await self._inner.latest(topic=topic) + + @overload + def producer(self, *, topic: StreamTopic[T]) -> StreamProducer[T]: ... + + @overload + def producer(self, *, topic: str | None = None) -> StreamProducer[Any]: ... + + def producer( + self, *, topic: str | StreamTopic[Any] | None = None + ) -> StreamProducer[Any]: + """A producer on ``topic`` that writes as this Activity attempt. + + The producer id is ``@`` and the + attempt is the Activity's Temporal attempt, so the first record of a + retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, + and the store deduplicates a repeated append within one attempt. + + Every call in one attempt returns the same producer for a topic, so + two parts of the Activity that each ask for it number their records + in one sequence. + """ + name, _ = self._inner._resolve(topic, None) + producer = self._producers.get(name) + if producer is None: + identity = self._identity + producer = self._producers[name] = StreamProducer( + self._inner, + name, + f"{identity.activity_id}@{identity.run_id}", + identity.attempt, + activity=identity, + ) + return producer + + +def activity_handle() -> ActivityStreamHandle: + """The stream of the Workflow that scheduled the running Activity. + + The handle is pinned to the scheduling run, so a read on it ends when + that run closes. To write to another Workflow's stream, use + :func:`get_stream_handle` with ``temporalio.activity.client()`` and a + producer id of your own. + + A Local Activity has no stream producer in this release. Its attempt is + not durable, so after a Worker crash it runs again as the same attempt, + and the store refuses its first append as stale. Use a regular Activity + instead. + + Raises: + ValueError: The Activity was not scheduled by a Workflow, or no stream + store is registered on the Worker's client. + StreamUnsupportedError: The Activity is a Local Activity. + RuntimeError: Not called from inside an Activity. + """ + info = temporalio.activity.info() + if info.is_local: + raise StreamUnsupportedError( + "a Local Activity has no stream producer in this release, since its " + "attempt isn't durable; use a regular Activity instead" + ) + if info.workflow_id is None or info.workflow_run_id is None: + raise ValueError( + "this Activity was not scheduled by a Workflow, so it has no Workflow " + "stream" + ) + client = temporalio.activity.client() + ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) + identity = ActivityProducer( + workflow_id=info.workflow_id, + run_id=info.workflow_run_id, + activity_id=info.activity_id, + attempt=info.attempt, + ) + inner = StreamHandle(store_for_client(client), client, ref) + return ActivityStreamHandle(inner, identity, _attempt_producers()) + + +# The producers of each running Activity attempt, by topic, under the id of the +# attempt's context. The context isn't hashable, so a finalizer drops the entry +# with it. +_activity_producers: dict[int, dict[str, StreamProducer[Any]]] = {} + + +def _attempt_producers() -> dict[str, StreamProducer[Any]]: + context = temporalio.activity._Context.current() + key = id(context) + producers = _activity_producers.get(key) + if producers is None: + producers = _activity_producers[key] = {} + weakref.finalize(context, _activity_producers.pop, key, None) + return producers + + +def get_stream_handle( + client: Client, + workflow_id: str | StreamRef, + *, + run_id: str | None = None, + topic: str | StreamTopic[Any] | None = None, +) -> StreamHandle: + """A handle on a Workflow's stream, through the client's store. + + Without ``run_id`` the handle follows the Workflow's run chain, so a + read continues across Continue-as-New and ends when the chain closes. + With one it is pinned to that run. ``topic`` is the topic a call that + names none addresses. + + Args: + client: A client that carries a stream store plugin. + workflow_id: The Workflow that owns the stream, or a + :class:`temporalio.contrib.streams.StreamRef` that names it. + run_id: Pin the handle to one run. + topic: The handle's default topic. Without one it is + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + + Raises: + ValueError: ``run_id`` or ``topic`` was given with a ref, or no + stream store is registered on ``client``. + """ + if isinstance(workflow_id, StreamRef): + if run_id is not None or topic is not None: + raise ValueError("a StreamRef carries its own run_id and topic") + ref = workflow_id + else: + ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) + return StreamHandle(store_for_client(client), client, ref) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py new file mode 100644 index 0000000000..cad6b4a6e0 --- /dev/null +++ b/tests/contrib/streams/test_outside_handles.py @@ -0,0 +1,450 @@ +"""Reaching a Workflow's stream from an Activity or a client. + +An Activity writes to the stream of the Workflow that scheduled it as +itself, so its retry is reported to readers as ``SUPERSEDED``. A client +reaches a Workflow's stream by Workflow id through its store. +""" + +from __future__ import annotations + +import dataclasses +import uuid +from collections.abc import Sequence +from datetime import timedelta + +import pytest + +from temporalio import activity, workflow +from temporalio.api.common.v1 import Payload +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.contrib.streams import ( + BEGINNING, + Cursor, + RecordKind, + StreamCursorError, + StreamNotFoundError, + StreamProducerError, + StreamRef, + StreamUnsupportedError, + Supersession, + activity_handle, + get_stream_handle, + topic, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.converter import ( + DataConverter, + PayloadCodec, + SerializationContext, + WithSerializationContext, +) +from temporalio.exceptions import ApplicationError +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id, read_all +from tests.helpers import new_worker + +EVENTS = topic("events", dict) +TOKENS = topic("tokens", str) + + +@workflow.defn +class Waits: + def __init__(self) -> None: + self.done = False + + @workflow.run + async def run(self) -> None: + await workflow.wait_condition(lambda: self.done) + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def running_owner(client: Client, task_queue: str) -> str: + # Core asks the server about the owner, so the stream's Workflow must exist. + workflow_id = new_workflow_id() + await client.start_workflow(Waits.run, id=workflow_id, task_queue=task_queue) + return workflow_id + + +@activity.defn +async def stream_tokens(tokens: list[str]) -> bool: + stream = activity_handle() + producer = stream.producer(topic=TOKENS) + attempt = activity.info().attempt + # The first attempt writes part of its answer and fails; the retry + # writes the whole answer, which readers must see as superseding it. + await producer.append(*tokens[:attempt]) + if attempt == 1: + raise ApplicationError("retry me") + await producer.append(*tokens[attempt:]) + await producer.finish() + return stream.ref.run_id == activity.info().workflow_run_id + + +@workflow.defn +class RunsStreamingActivity: + @workflow.run + async def run(self, tokens: list[str]) -> bool: + return await workflow.execute_activity( + stream_tokens, + tokens, + activity_id="tokens", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=50), maximum_attempts=2 + ), + ) + + +async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsStreamingActivity, activities=[stream_tokens] + ) as worker: + pinned = await streams_client.execute_workflow( + RunsStreamingActivity.run, + ["a", "b", "c"], + id=workflow_id, + task_queue=worker.task_queue, + ) + assert pinned + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.kind for r in records] == [ + RecordKind.DATA, + RecordKind.SUPERSEDED, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.DATA, + RecordKind.FINISH, + ] + described = await streams_client.get_workflow_handle(workflow_id).describe() + # Core derives the same producer id the handle reports. + producer_id = f"tokens@{described.run_id}" + assert records[1].supersession == Supersession(producer_id, 1, 2) + assert [r.value for r in records if r.kind is RecordKind.DATA] == [ + "a", + "a", + "b", + "c", + ] + assert [(r.producer_id, r.attempt, r.sequence) for r in records[2:]] == [ + (producer_id, 2, 1), + (producer_id, 2, 2), + (producer_id, 2, 3), + (producer_id, 2, 4), + ] + + +@activity.defn +async def stream_without_store() -> str: + try: + activity_handle() + except ValueError as error: + return str(error) + return "opened" + + +@workflow.defn +class RunsActivityWithoutStore: + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity( + stream_without_store, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_without_a_store_is_told(client: Client): + async with new_worker( + client, RunsActivityWithoutStore, activities=[stream_without_store] + ) as worker: + said = await client.execute_workflow( + RunsActivityWithoutStore.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "no stream store is registered" in said + + +async def test_a_client_handle_follows_the_chain_or_pins_a_run(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + follower = get_stream_handle(streams_client, "wf") + assert follower.ref == StreamRef.for_workflow("wf") + pinned = get_stream_handle(streams_client, "wf", run_id="run", topic=EVENTS) + assert pinned.ref == StreamRef.for_workflow("wf", run_id="run", topic=EVENTS) + assert get_stream_handle(streams_client, pinned.ref).ref == pinned.ref + with pytest.raises(ValueError, match="carries its own"): + get_stream_handle(streams_client, pinned.ref, run_id="other") + with pytest.raises(ValueError, match="no stream store is registered"): + get_stream_handle(client, "wf") + + +async def test_a_producer_retry_dedupes_and_a_divergent_one_is_refused( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + assert await stream.latest() == BEGINNING + producer = stream.producer(producer_id="p", attempt=1) + assert await producer.append() == BEGINNING + first = await producer.append("a", "b") + assert await producer.append() == first + assert await stream.latest() == first + # A second object for the same session starts at one again: a repeat + # of the newest batch dedupes to the original position. + again = stream.producer(producer_id="p", attempt=1) + assert await again.append("a", "b") == first + divergent = stream.producer(producer_id="p", attempt=1) + with pytest.raises(StreamProducerError): + await divergent.append("x", "y") + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + records = await read_all(stream.read()) + assert [r.value for r in records] == ["a", "b"] + + +async def test_a_resumed_read_continues_after_its_cursor(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, Waits) as worker: + workflow_id = await running_owner(streams_client, worker.task_queue) + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + producer = stream.producer(producer_id="p", attempt=1) + middle = await producer.append("a", "b") + await producer.append("c") + # A read from END starts at its first iteration, so a reader that must + # not miss the next record positions itself with latest() first. + live = stream.read(after=await stream.latest()) + await producer.append("d") + assert (await anext(live)).value == "d" + await live.aclose() + with pytest.raises(StreamCursorError): + await anext(stream.read(after=Cursor("not a cursor"))) + owner = streams_client.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read(after=middle))] == ["c", "d"] + + +async def test_a_stream_without_an_owner_is_not_found(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + stream = get_stream_handle(streams_client, f"no-such-owner-{uuid.uuid4()}") + with pytest.raises(StreamNotFoundError): + await stream.producer(producer_id="p", attempt=1).append("a") + + +async def test_a_producer_refuses_a_bad_identity(client: Client): + stream = get_stream_handle(await connect_with(client, MemoryStreams()), "wf") + with pytest.raises(ValueError, match="producer_id"): + stream.producer(producer_id="", attempt=1) + for attempt in (0, True, 1.5): + with pytest.raises(ValueError, match="attempt"): + stream.producer(producer_id="p", attempt=attempt) # type: ignore[arg-type] + + +class NonceCodec(PayloadCodec): + """Encrypts with a fresh nonce per call, so two encodings never match.""" + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/nonce"}, + data=uuid.uuid4().bytes + p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [Payload.FromString(p.data[16:]) for p in payloads] + + +async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): + coded = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=NonceCodec()) + ) + async with new_worker(coded, Waits) as worker: + workflow_id = await running_owner(coded, worker.task_queue) + stream = get_stream_handle(coded, workflow_id, topic=TOKENS) + first = await stream.producer(producer_id="p", attempt=1).append("a") + retry = stream.producer(producer_id="p", attempt=1) + # The ciphertext differs, and the digest lang took before the codec does not. + assert await retry.append("a") == first + owner = coded.get_workflow_handle(workflow_id) + await owner.signal(Waits.finish) + await owner.result() + assert [r.value for r in await read_all(stream.read())] == ["a"] + + +async def test_an_activity_with_no_workflow_has_no_workflow_stream(): + env = ActivityEnvironment() + env.info = dataclasses.replace( + ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None + ) + + async def standalone() -> None: + activity_handle() + + with pytest.raises(ValueError, match="not scheduled by a Workflow"): + await env.run(standalone) + + +@activity.defn +async def count_twice() -> str: + producer = activity_handle().producer(topic=TOKENS) + await producer.append("one") + await producer.append("two") + return producer.producer_id + + +@workflow.defn +class CountThenContinue: + @workflow.run + async def run(self, runs_left: int) -> list[str]: + # The same Activity id in every run, as a counter-based id would be. + producer_id = await workflow.execute_activity( + count_twice, + activity_id="count", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=1), + ) + if runs_left: + workflow.continue_as_new(runs_left - 1) + return [producer_id] + + +async def test_an_activity_producer_writes_in_every_run_of_a_chain(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, CountThenContinue, activities=[count_twice] + ) as worker: + handle = await streams_client.start_workflow( + CountThenContinue.run, 1, id=workflow_id, task_queue=worker.task_queue + ) + (last_producer,) = await handle.result() + stream = get_stream_handle(streams_client, workflow_id, topic=TOKENS) + records = await read_all(stream.read()) + # Both runs' records, under two producers: the run is part of the id. + assert [r.value for r in records] == ["one", "two", "one", "two"] + producers = [r.producer_id for r in records] + assert producers[0] == producers[1] != producers[2] == producers[3] + assert producers[3] == last_producer + assert last_producer.endswith("@" + (await handle.describe()).run_id) + + +class WorkflowKeyedCodec(PayloadCodec, WithSerializationContext): + """Encrypts each payload under the key of the Workflow its context names, as + a multi-tenant codec does. A payload decodes only under the same key.""" + + def __init__(self, key: str = "") -> None: + self.key = key + + def with_context(self, context: SerializationContext) -> WorkflowKeyedCodec: + workflow_id = getattr(context, "workflow_id", None) + return WorkflowKeyedCodec(workflow_id or "") + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + return [ + Payload( + metadata={"encoding": b"binary/keyed", "key": self.key.encode()}, + data=p.SerializeToString(), + ) + for p in payloads + ] + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + out = [] + for p in payloads: + if p.metadata.get("encoding") != b"binary/keyed": + out.append(p) + continue + if p.metadata["key"].decode() != self.key: + raise ValueError( + f"encrypted under {p.metadata['key']!r}, not {self.key!r}" + ) + out.append(Payload.FromString(p.data)) + return out + + +@activity.defn +async def append_one_token() -> None: + await activity_handle().producer(topic=TOKENS).append("from the activity") + + +@workflow.defn +class PublishesThenRunsAnActivity: + @workflow.run + async def run(self) -> None: + await workflow.execute_activity( + append_one_token, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_both_sides_use_the_workflows_serialization_context(client: Client): + # The stream is the Workflow's, so its producers and readers encode with + # the Workflow's context. + keyed = await connect_with( + client, MemoryStreams(), DataConverter(payload_codec=WorkflowKeyedCodec()) + ) + workflow_id = new_workflow_id() + async with new_worker( + keyed, PublishesThenRunsAnActivity, activities=[append_one_token] + ) as worker: + await keyed.execute_workflow( + PublishesThenRunsAnActivity.run, + id=workflow_id, + task_queue=worker.task_queue, + ) + records = await read_all(get_stream_handle(keyed, workflow_id).read(topic=TOKENS)) + assert [r.value for r in records] == ["from the activity"] + + +@activity.defn +async def logs_twice() -> None: + # A helper that asks for the producer on every call. + for line in ["started", "started"]: + await activity_handle().producer(topic=TOKENS).append(line) + + +@workflow.defn +class RunsLogsTwice: + @workflow.run + async def run(self) -> None: + await workflow.execute_activity( + logs_twice, start_to_close_timeout=timedelta(seconds=10) + ) + + +async def test_an_activity_gets_one_producer_per_topic(client: Client): + # Two producer objects would both start at sequence 1, so the second equal + # record would dedupe into the first and vanish. + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker( + streams_client, RunsLogsTwice, activities=[logs_twice] + ) as worker: + await streams_client.execute_workflow( + RunsLogsTwice.run, id=workflow_id, task_queue=worker.task_queue + ) + records = await read_all( + get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) + ) + assert [r.value for r in records] == ["started", "started"] + + +async def test_a_local_activity_has_no_stream_producer(): + env = ActivityEnvironment() + env.info = dataclasses.replace(ActivityEnvironment.default_info(), is_local=True) + + async def local() -> None: + activity_handle() + + with pytest.raises(StreamUnsupportedError, match="Local Activity"): + await env.run(local) From 6527082bb2fa1ad3695bc6694cdae2a6f790d5bb Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:21:30 -0700 Subject: [PATCH 2/5] Refused stream owners this release does not support. --- temporalio/contrib/streams/__init__.py | 6 ++ temporalio/contrib/streams/_handles.py | 14 ++-- temporalio/contrib/streams/_ref.py | 32 ++++++-- temporalio/contrib/streams/_workflow.py | 30 ++++++++ tests/contrib/streams/test_outside_handles.py | 13 ---- tests/contrib/streams/test_topics_refs.py | 19 +++-- .../streams/test_unsupported_owners.py | 73 +++++++++++++++++++ 7 files changed, 152 insertions(+), 35 deletions(-) create mode 100644 temporalio/contrib/streams/_workflow.py create mode 100644 tests/contrib/streams/test_unsupported_owners.py diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index c6976630bb..70d9e4d175 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -43,6 +43,10 @@ :func:`get_stream_handle`, and writes with a producer id and an attempt of its own. +In this release only a Workflow owns a stream, and only Activities and +clients read one. Reading inside a Workflow (:func:`workflow_reader`) and +the other owner kinds raise :class:`StreamUnsupportedError`. + The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` as an ordinary payload. @@ -78,6 +82,7 @@ ) from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic +from temporalio.contrib.streams._workflow import workflow_reader __all__ = [ "ActivityStreamHandle", @@ -108,4 +113,5 @@ "activity_handle", "get_stream_handle", "topic", + "workflow_reader", ] diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index e922484e30..4c43eaa69d 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -490,9 +490,10 @@ def activity_handle() -> ActivityStreamHandle: instead. Raises: - ValueError: The Activity was not scheduled by a Workflow, or no stream - store is registered on the Worker's client. - StreamUnsupportedError: The Activity is a Local Activity. + StreamUnsupportedError: The Activity is a Local Activity, or it was + not scheduled by a Workflow, since streams owned by an Activity + are not supported in this release. + ValueError: No stream store is registered on the Worker's client. RuntimeError: Not called from inside an Activity. """ info = temporalio.activity.info() @@ -502,9 +503,9 @@ def activity_handle() -> ActivityStreamHandle: "attempt isn't durable; use a regular Activity instead" ) if info.workflow_id is None or info.workflow_run_id is None: - raise ValueError( - "this Activity was not scheduled by a Workflow, so it has no Workflow " - "stream" + raise StreamUnsupportedError( + "this Activity was not scheduled by a Workflow, and streams owned by " + "an Activity are not supported in this release" ) client = temporalio.activity.client() ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) @@ -566,4 +567,5 @@ def get_stream_handle( ref = workflow_id else: ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) + ref._require_supported() return StreamHandle(store_for_client(client), client, ref) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index bc8308398a..91f78a8c4e 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -13,6 +13,7 @@ from dataclasses import dataclass from typing import Any, Literal +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._topic import ( DEFAULT_TOPIC, StreamTopic, @@ -25,8 +26,7 @@ """The owner kinds this release opens. Only a Workflow owns a stream. :attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind -a later release adds still reaches the ref's own check when it is decoded, -instead of failing the converter's type check. +a later release adds still decodes, and is refused where a handle opens. """ @@ -53,19 +53,35 @@ class StreamRef: topic: str = DEFAULT_TOPIC def __post_init__(self) -> None: - """Refuse a ref this release cannot open. + """Refuse a ref that names no stream. + + Any ``kind`` decodes, since a ref of a later release can arrive as + Workflow input, where raising would fail the Workflow Task on every + retry. :meth:`_require_supported` refuses it where a handle opens. Raises: - ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or - ``topic`` is empty. + ValueError: ``kind`` or ``topic`` is empty, or a Workflow ref has + no ``workflow_id``. """ - if self.kind != "workflow": - raise ValueError(f"unknown StreamRef kind {self.kind!r}") - if not self.workflow_id: + if not self.kind: + raise ValueError("a StreamRef needs a kind") + if self.kind == "workflow" and not self.workflow_id: raise ValueError("a Workflow StreamRef needs a workflow_id") if not self.topic: raise ValueError("a StreamRef needs a topic name") + def _require_supported(self) -> None: + """Refuse to open a stream whose owner kind this release lacks. + + Raises: + StreamUnsupportedError: ``kind`` is not ``"workflow"``. + """ + if self.kind != "workflow": + raise StreamUnsupportedError( + f"streams owned by kind {self.kind!r} are not supported in this " + "release; only Workflow-owned streams are" + ) + @classmethod def for_workflow( cls, diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py new file mode 100644 index 0000000000..8c371a4c54 --- /dev/null +++ b/temporalio/contrib/streams/_workflow.py @@ -0,0 +1,30 @@ +"""Streams inside Workflow code. + +Reading a stream inside a Workflow is not part of this release. +""" + +from __future__ import annotations + +from typing import Any + +from temporalio.contrib.streams._errors import StreamUnsupportedError + +__all__ = ["workflow_reader"] + + +def workflow_reader(*_args: Any, **_kwargs: Any) -> Any: + """Reading a stream inside a Workflow is not supported in this release. + + Read from an Activity or a client with + :func:`temporalio.contrib.streams.activity_handle` or + :func:`temporalio.contrib.streams.get_stream_handle`, and send the + Workflow what it needs, for example as a Signal. + + Raises: + StreamUnsupportedError: Always. + """ + raise StreamUnsupportedError( + "reading a stream inside a Workflow is not supported in this release; " + "read it from an Activity or a client and send the Workflow what it " + "needs, for example as a Signal" + ) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index cad6b4a6e0..b466aa8ca6 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -282,19 +282,6 @@ async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): assert [r.value for r in await read_all(stream.read())] == ["a"] -async def test_an_activity_with_no_workflow_has_no_workflow_stream(): - env = ActivityEnvironment() - env.info = dataclasses.replace( - ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None - ) - - async def standalone() -> None: - activity_handle() - - with pytest.raises(ValueError, match="not scheduled by a Workflow"): - await env.run(standalone) - - @activity.defn async def count_twice() -> str: producer = activity_handle().producer(topic=TOKENS) diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 3cbbe1e6b0..509d8cbbf1 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -9,6 +9,7 @@ DEFAULT_TOPIC, StreamRef, StreamTopic, + StreamUnsupportedError, topic, ) from temporalio.contrib.streams._topic import ( @@ -51,24 +52,26 @@ def test_a_ref_round_trips_through_the_default_converter(): def test_a_ref_refuses_what_it_cannot_name(): - with pytest.raises(ValueError, match="unknown"): - StreamRef("nexus", "x") with pytest.raises(ValueError): StreamRef("workflow", "") with pytest.raises(ValueError): StreamRef("workflow", "wf", topic="") -def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): - # The kind is a plain string, so the converter builds the ref and the ref - # itself refuses the kind, rather than the converter failing a type check. +@pytest.mark.parametrize("kind", ["activity", "nexus"]) +def test_a_ref_from_a_later_release_decodes_and_fails_where_it_opens(kind: str): + # Decoding happens while a Workflow Task applies its input, where a raise + # fails the task on every retry. Opening the handle is where the caller + # can catch it. converter = DataConverter.default.payload_converter payload = Payload( metadata={"encoding": b"json/plain"}, - data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + data=f'{{"kind": "{kind}", "workflow_id": "wf", "topic": "out"}}'.encode(), ) - with pytest.raises(ValueError, match="'activity'"): - converter.from_payloads([payload], [StreamRef]) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref.kind == kind + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + ref._require_supported() @pytest.mark.parametrize( diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py new file mode 100644 index 0000000000..498957f430 --- /dev/null +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -0,0 +1,73 @@ +"""Owners and readers this release does not support say so. + +Only a Workflow owns a stream in this release, and only Activities and +clients read one. Everything else raises ``StreamUnsupportedError``. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.streams import ( + StreamRef, + StreamUnsupportedError, + activity_handle, + get_stream_handle, + topic, + workflow_reader, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.testing import ActivityEnvironment +from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.helpers import new_worker + +EVENTS = topic("events", dict) + + +async def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens( + client: Client, +): + streams_client = await connect_with(client, MemoryStreams()) + for kind in ("activity", "standalone"): + ref = StreamRef(kind, "x") # type: ignore[arg-type] + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + get_stream_handle(streams_client, ref) + + +@workflow.defn +class TriesToRead: + @workflow.run + async def run(self) -> str: + try: + workflow_reader(EVENTS) + except StreamUnsupportedError as error: + return str(error) + return "read" + + +async def test_a_workflow_cannot_read_a_stream(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + async with new_worker(streams_client, TriesToRead) as worker: + said = await streams_client.execute_workflow( + TriesToRead.run, + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + assert "not supported in this release" in said + + +async def test_an_activity_with_no_workflow_has_no_stream(): + env = ActivityEnvironment() + env.info = dataclasses.replace( + ActivityEnvironment.default_info(), workflow_id=None, workflow_run_id=None + ) + + async def standalone() -> None: + activity_handle() + + with pytest.raises(StreamUnsupportedError, match="owned by an Activity"): + await env.run(standalone) From 09236ce99dc188a914d7f601b7f28d468bcd5f0d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 17:52:04 -0700 Subject: [PATCH 3/5] Said that SUPERSEDED withdraws one producer's records, not others'. --- temporalio/contrib/streams/_handles.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index e922484e30..161559bfca 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -263,7 +263,9 @@ def read( it back sees every stored record once. A resumed read knows the attempt of the record at its cursor, so a new attempt of that producer still arrives as ``SUPERSEDED``. Earlier attempts of other - producers are not known to it. The read ends when the owner's run + producers are not known to it. A ``SUPERSEDED`` record withdraws only + what its producer wrote in earlier attempts. The other producers' + records on the topic stay valid. The read ends when the owner's run chain is closed and every retained record has been delivered. The result is a generator, so a caller that stops early calls ``aclose()`` on it. @@ -345,8 +347,9 @@ def producer( """A producer on ``topic`` that writes as ``producer_id`` in ``attempt``. A process that restarts its work raises ``attempt``, which starts a - new sequence and tells readers that what the earlier attempt wrote is - superseded. Keep one producer object per ``(topic, producer_id, + new sequence and tells readers that what the earlier attempt of this + ``producer_id`` wrote is superseded. Records of other producers are + not. Keep one producer object per ``(topic, producer_id, attempt)``. Each object numbers its own records from one. A second object for the same session is refused as stale once the first moved past its first batch. Before that, a second object's first batch with @@ -455,8 +458,10 @@ def producer( The producer id is ``@`` and the attempt is the Activity's Temporal attempt, so the first record of a - retry makes readers see ``SUPERSEDED``. A retry starts a new sequence, - and the store deduplicates a repeated append within one attempt. + retry makes readers see ``SUPERSEDED``, which withdraws what this + Activity's earlier attempts wrote and nothing else. A retry starts a + new sequence, and the store deduplicates a repeated append within one + attempt. Every call in one attempt returns the same producer for a topic, so two parts of the Activity that each ask for it number their records From b9d2eb65313bb624ba1feea74d55af6cd9d10caf Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 17:53:22 -0700 Subject: [PATCH 4/5] Removed run pinning from stream handles, so each read follows the chain. --- temporalio/contrib/streams/__init__.py | 6 ++- temporalio/contrib/streams/_handles.py | 52 +++++++++++-------- tests/contrib/streams/test_outside_handles.py | 34 ++++++++---- 3 files changed, 59 insertions(+), 33 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index 6c2478cb2e..aa42a02e58 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -39,12 +39,14 @@ - An Activity reaches the stream of the Workflow that scheduled it with :func:`activity_handle`, and writes as itself: its Activity id with the - scheduling run id, and its Temporal attempt. The handle is pinned to that - run, so a read on it ends when the run closes, Continue-as-New included. + scheduling run id, and its Temporal attempt. - Any process holding a client reaches a Workflow's stream with :func:`get_stream_handle`, and writes with a producer id and an attempt of its own. +Every read follows the Workflow's run chain across Continue-as-New and ends +when the chain closes. A reader can't pin one run. + The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` as an ordinary payload. diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 161559bfca..455b735cec 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -160,7 +160,7 @@ async def _write(self, kind: RecordKind, values: Any) -> Cursor: for index, value in enumerate(values) ] request = AppendRequest( - stream=self._handle._address(self._topic), + stream=self._handle._address(self._topic, producer=True), sequence=self._sequence, records=await _append_records(converter, wires), digest=batch_digest(wires), @@ -197,20 +197,28 @@ async def _append_records( class StreamHandle: """One stream, addressed by topic, from outside Workflow code. - A handle on a Workflow's stream follows the run chain unless its - :attr:`ref` is pinned to a run. A topic is a + A handle on a Workflow's stream follows the run chain, so a read goes on + across Continue-as-New and ends when the chain closes. A topic is a :class:`temporalio.contrib.streams.StreamTopic`, which carries the record type, or a plain string with ``result_type=`` for a name decided at runtime. A call that names no topic addresses the topic of :attr:`ref`. """ def __init__( - self, plugin: StreamStorePlugin, client: Client, ref: StreamRef + self, + plugin: StreamStorePlugin, + client: Client, + ref: StreamRef, + *, + _producer_run_id: str = "", ) -> None: """Prefer :func:`temporalio.contrib.streams.get_stream_handle`.""" self._plugin = plugin self._client = client self._ref = ref + # An Activity writes to the chain of the run that scheduled it, even + # once a new chain took the Workflow id. Reads follow the chain. + self._producer_run_id = _producer_run_id # The stream is the Workflow's, and the Workflow encodes its records with # its own context, so every reader and producer on it uses that context. self._converter: DataConverter = client.data_converter.with_context( @@ -371,12 +379,12 @@ def _resolve( ) -> tuple[str, type | None]: return resolve_topic(self._ref.topic if topic is None else topic, result_type) - def _address(self, topic: str) -> StreamAddress: + def _address(self, topic: str, *, producer: bool = False) -> StreamAddress: return StreamAddress( namespace=self._client.namespace, owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, workflow_id=self._ref.workflow_id, - run_id=self._ref.run_id or "", + run_id=self._producer_run_id if producer else "", topic=topic, ) @@ -403,7 +411,7 @@ def __init__( @property def ref(self) -> StreamRef: - """The stream this handle is on, pinned to the scheduling run.""" + """The stream this handle is on, as data another process can open.""" return self._inner.ref @overload @@ -484,8 +492,9 @@ def producer( def activity_handle() -> ActivityStreamHandle: """The stream of the Workflow that scheduled the running Activity. - The handle is pinned to the scheduling run, so a read on it ends when - that run closes. To write to another Workflow's stream, use + A read on it follows the Workflow's run chain, as every read does. Its + producers write to the chain of the scheduling run. To write to another + Workflow's stream, use :func:`get_stream_handle` with ``temporalio.activity.client()`` and a producer id of your own. @@ -512,14 +521,16 @@ def activity_handle() -> ActivityStreamHandle: "stream" ) client = temporalio.activity.client() - ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) + ref = StreamRef.for_workflow(info.workflow_id) identity = ActivityProducer( workflow_id=info.workflow_id, run_id=info.workflow_run_id, activity_id=info.activity_id, attempt=info.attempt, ) - inner = StreamHandle(store_for_client(client), client, ref) + inner = StreamHandle( + store_for_client(client), client, ref, _producer_run_id=info.workflow_run_id + ) return ActivityStreamHandle(inner, identity, _attempt_producers()) @@ -543,32 +554,29 @@ def get_stream_handle( client: Client, workflow_id: str | StreamRef, *, - run_id: str | None = None, topic: str | StreamTopic[Any] | None = None, ) -> StreamHandle: """A handle on a Workflow's stream, through the client's store. - Without ``run_id`` the handle follows the Workflow's run chain, so a - read continues across Continue-as-New and ends when the chain closes. - With one it is pinned to that run. ``topic`` is the topic a call that - names none addresses. + The handle follows the Workflow's run chain, so a read continues across + Continue-as-New and ends when the chain closes. ``topic`` is the topic a + call that names none addresses. Args: client: A client that carries a stream store plugin. workflow_id: The Workflow that owns the stream, or a :class:`temporalio.contrib.streams.StreamRef` that names it. - run_id: Pin the handle to one run. topic: The handle's default topic. Without one it is :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. Raises: - ValueError: ``run_id`` or ``topic`` was given with a ref, or no - stream store is registered on ``client``. + ValueError: ``topic`` was given with a ref, or no stream store is + registered on ``client``. """ if isinstance(workflow_id, StreamRef): - if run_id is not None or topic is not None: - raise ValueError("a StreamRef carries its own run_id and topic") + if topic is not None: + raise ValueError("a StreamRef carries its own topic") ref = workflow_id else: - ref = StreamRef.for_workflow(workflow_id, run_id=run_id, topic=topic) + ref = StreamRef.for_workflow(workflow_id, topic=topic) return StreamHandle(store_for_client(client), client, ref) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index cad6b4a6e0..d5747b7657 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -81,7 +81,7 @@ async def stream_tokens(tokens: list[str]) -> bool: raise ApplicationError("retry me") await producer.append(*tokens[attempt:]) await producer.finish() - return stream.ref.run_id == activity.info().workflow_run_id + return stream.ref == StreamRef.for_workflow(activity.info().workflow_id or "") @workflow.defn @@ -105,13 +105,13 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): async with new_worker( streams_client, RunsStreamingActivity, activities=[stream_tokens] ) as worker: - pinned = await streams_client.execute_workflow( + follows_chain = await streams_client.execute_workflow( RunsStreamingActivity.run, ["a", "b", "c"], id=workflow_id, task_queue=worker.task_queue, ) - assert pinned + assert follows_chain records = await read_all( get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) ) @@ -171,15 +171,16 @@ async def test_an_activity_without_a_store_is_told(client: Client): assert "no stream store is registered" in said -async def test_a_client_handle_follows_the_chain_or_pins_a_run(client: Client): +async def test_a_client_handle_follows_the_chain(client: Client): streams_client = await connect_with(client, MemoryStreams()) follower = get_stream_handle(streams_client, "wf") assert follower.ref == StreamRef.for_workflow("wf") - pinned = get_stream_handle(streams_client, "wf", run_id="run", topic=EVENTS) - assert pinned.ref == StreamRef.for_workflow("wf", run_id="run", topic=EVENTS) - assert get_stream_handle(streams_client, pinned.ref).ref == pinned.ref - with pytest.raises(ValueError, match="carries its own"): - get_stream_handle(streams_client, pinned.ref, run_id="other") + assert follower._address("output").run_id == "" + events = get_stream_handle(streams_client, "wf", topic=EVENTS) + assert events.ref == StreamRef.for_workflow("wf", topic=EVENTS) + assert get_stream_handle(streams_client, events.ref).ref == events.ref + with pytest.raises(ValueError, match="carries its own topic"): + get_stream_handle(streams_client, events.ref, topic=TOKENS) with pytest.raises(ValueError, match="no stream store is registered"): get_stream_handle(client, "wf") @@ -282,6 +283,21 @@ async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): assert [r.value for r in await read_all(stream.read())] == ["a"] +async def test_an_activity_reads_the_chain_and_writes_to_its_run(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + env = ActivityEnvironment(client=streams_client) + + async def addresses() -> tuple[str, str]: + inner = activity_handle()._inner + return ( + inner._address("output").run_id, + inner._address("output", producer=True).run_id, + ) + + read, write = await env.run(addresses) + assert (read, write) == ("", ActivityEnvironment.default_info().workflow_run_id) + + async def test_an_activity_with_no_workflow_has_no_workflow_stream(): env = ActivityEnvironment() env.info = dataclasses.replace( From 59bc227da59b8f26705c09160841ce0ac5acfe61 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 18:19:46 -0700 Subject: [PATCH 5/5] Let a handle name its chain by a run, and read it to the chain's end. --- temporalio/contrib/streams/_handles.py | 84 +++++++++--- tests/contrib/streams/test_outside_handles.py | 120 ++++++++++++++++-- 2 files changed, 172 insertions(+), 32 deletions(-) diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 455b735cec..3eec3a163b 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -38,7 +38,11 @@ from temporalio.client import Client from temporalio.contrib.streams._body import batch_digest, content_hash, encode_bodies from temporalio.contrib.streams._cursor import BEGINNING -from temporalio.contrib.streams._errors import StreamUnsupportedError +from temporalio.contrib.streams._errors import ( + StreamError, + StreamNotFoundError, + StreamUnsupportedError, +) from temporalio.contrib.streams._plugin import ( StreamStorePlugin, call, @@ -49,6 +53,7 @@ from temporalio.contrib.streams._topic import StreamTopic, resolve_topic from temporalio.contrib.streams._wire import WireRecord, from_read, to_wire from temporalio.converter import DataConverter, WorkflowSerializationContext +from temporalio.service import RPCError, RPCStatusCode __all__ = [ "ActivityStreamHandle", @@ -160,7 +165,7 @@ async def _write(self, kind: RecordKind, values: Any) -> Cursor: for index, value in enumerate(values) ] request = AppendRequest( - stream=self._handle._address(self._topic, producer=True), + stream=self._handle._address(self._topic), sequence=self._sequence, records=await _append_records(converter, wires), digest=batch_digest(wires), @@ -209,16 +214,11 @@ def __init__( plugin: StreamStorePlugin, client: Client, ref: StreamRef, - *, - _producer_run_id: str = "", ) -> None: """Prefer :func:`temporalio.contrib.streams.get_stream_handle`.""" self._plugin = plugin self._client = client self._ref = ref - # An Activity writes to the chain of the run that scheduled it, even - # once a new chain took the Workflow id. Reads follow the chain. - self._producer_run_id = _producer_run_id # The stream is the Workflow's, and the Workflow encodes its records with # its own context, so every reader and producer on it uses that context. self._converter: DataConverter = client.data_converter.with_context( @@ -312,7 +312,8 @@ async def _read( self, topic: str, after: Cursor, result_type: type | None ) -> AsyncGenerator[StreamRecord[Any], None]: service = await self._plugin._service_for(self._client) - request = ReadRequest(stream=self._address(topic), after=after.token) + address = self._address(topic, run_id=await self._read_run_id()) + request = ReadRequest(stream=address, after=after.token) request.wait.FromTimedelta(_READ_WAIT) while True: response = await call(service.read(request)) @@ -379,15 +380,50 @@ def _resolve( ) -> tuple[str, type | None]: return resolve_topic(self._ref.topic if topic is None else topic, result_type) - def _address(self, topic: str, *, producer: bool = False) -> StreamAddress: + def _address(self, topic: str, *, run_id: str | None = None) -> StreamAddress: return StreamAddress( namespace=self._client.namespace, owner_kind=StreamOwnerKind.STREAM_OWNER_KIND_WORKFLOW, workflow_id=self._ref.workflow_id, - run_id=self._producer_run_id if producer else "", + run_id=(self._ref.chain_run_id or "") if run_id is None else run_id, topic=topic, ) + async def _read_run_id(self) -> str: + """The run a read names to Core, which ends a read when that run closes. + + A read must end only with its chain. While the chain that holds + ``chain_run_id`` is the Workflow id's newest, the read names no run + and follows that chain to its end. Once a newer chain took the id, + the old chain has ended, and a read on any of its runs delivers what + is left and ends. + + Raises: + StreamNotFoundError: Temporal holds no run ``chain_run_id``. + StreamError: Describing the Workflow failed. + """ + run_id = self._ref.chain_run_id + if not run_id: + return "" + workflow_id = self._ref.workflow_id + try: + named = await self._client.get_workflow_handle( + workflow_id, run_id=run_id + ).describe() + newest = await self._client.get_workflow_handle(workflow_id).describe() + except RPCError as error: + if error.status == RPCStatusCode.NOT_FOUND: + raise StreamNotFoundError( + f"Workflow {workflow_id!r} has no run {run_id!r}" + ) from error + raise StreamError( + f"could not describe Workflow {workflow_id!r}: {error}" + ) from error + first = named.raw_description.workflow_execution_info.first_run_id + if first == newest.raw_description.workflow_execution_info.first_run_id: + return "" + return run_id + class ActivityStreamHandle: """The scheduling Workflow's stream, as the running Activity sees it. @@ -492,8 +528,10 @@ def producer( def activity_handle() -> ActivityStreamHandle: """The stream of the Workflow that scheduled the running Activity. - A read on it follows the Workflow's run chain, as every read does. Its - producers write to the chain of the scheduling run. To write to another + Its ref names the chain of the scheduling run with ``chain_run_id``, so + its producers write to that chain even once a new chain took the + Workflow id. A read on it follows that chain across Continue-as-New and + ends when the chain ends. To write to another Workflow's stream, use :func:`get_stream_handle` with ``temporalio.activity.client()`` and a producer id of your own. @@ -521,16 +559,14 @@ def activity_handle() -> ActivityStreamHandle: "stream" ) client = temporalio.activity.client() - ref = StreamRef.for_workflow(info.workflow_id) + ref = StreamRef.for_workflow(info.workflow_id, chain_run_id=info.workflow_run_id) identity = ActivityProducer( workflow_id=info.workflow_id, run_id=info.workflow_run_id, activity_id=info.activity_id, attempt=info.attempt, ) - inner = StreamHandle( - store_for_client(client), client, ref, _producer_run_id=info.workflow_run_id - ) + inner = StreamHandle(store_for_client(client), client, ref) return ActivityStreamHandle(inner, identity, _attempt_producers()) @@ -555,6 +591,7 @@ def get_stream_handle( workflow_id: str | StreamRef, *, topic: str | StreamTopic[Any] | None = None, + chain_run_id: str | None = None, ) -> StreamHandle: """A handle on a Workflow's stream, through the client's store. @@ -568,15 +605,20 @@ def get_stream_handle( :class:`temporalio.contrib.streams.StreamRef` that names it. topic: The handle's default topic. Without one it is :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + chain_run_id: Any run of the chain to open. Without one the handle + opens the Workflow id's newest chain. It does not end a read + when that run closes. Raises: - ValueError: ``topic`` was given with a ref, or no stream store is - registered on ``client``. + ValueError: ``topic`` or ``chain_run_id`` was given with a ref, or no + stream store is registered on ``client``. """ if isinstance(workflow_id, StreamRef): - if topic is not None: - raise ValueError("a StreamRef carries its own topic") + if topic is not None or chain_run_id is not None: + raise ValueError("a StreamRef carries its own topic and chain_run_id") ref = workflow_id else: - ref = StreamRef.for_workflow(workflow_id, topic=topic) + ref = StreamRef.for_workflow( + workflow_id, topic=topic, chain_run_id=chain_run_id + ) return StreamHandle(store_for_client(client), client, ref) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index d5747b7657..3aed14d934 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import dataclasses import uuid from collections.abc import Sequence @@ -81,7 +82,10 @@ async def stream_tokens(tokens: list[str]) -> bool: raise ApplicationError("retry me") await producer.append(*tokens[attempt:]) await producer.finish() - return stream.ref == StreamRef.for_workflow(activity.info().workflow_id or "") + info = activity.info() + return stream.ref == StreamRef.for_workflow( + info.workflow_id or "", chain_run_id=info.workflow_run_id + ) @workflow.defn @@ -105,13 +109,13 @@ async def test_an_activity_retry_supersedes_its_first_attempt(client: Client): async with new_worker( streams_client, RunsStreamingActivity, activities=[stream_tokens] ) as worker: - follows_chain = await streams_client.execute_workflow( + names_its_chain = await streams_client.execute_workflow( RunsStreamingActivity.run, ["a", "b", "c"], id=workflow_id, task_queue=worker.task_queue, ) - assert follows_chain + assert names_its_chain records = await read_all( get_stream_handle(streams_client, workflow_id).read(topic=TOKENS) ) @@ -181,6 +185,12 @@ async def test_a_client_handle_follows_the_chain(client: Client): assert get_stream_handle(streams_client, events.ref).ref == events.ref with pytest.raises(ValueError, match="carries its own topic"): get_stream_handle(streams_client, events.ref, topic=TOKENS) + named = get_stream_handle(streams_client, "wf", chain_run_id="run") + assert named.ref == StreamRef.for_workflow("wf", chain_run_id="run") + # Appends and positions name the chain's run. A read decides at its start. + assert named._address("output").run_id == "run" + with pytest.raises(ValueError, match="chain_run_id"): + get_stream_handle(streams_client, named.ref, chain_run_id="other") with pytest.raises(ValueError, match="no stream store is registered"): get_stream_handle(client, "wf") @@ -283,19 +293,107 @@ async def test_a_retry_through_a_nonce_codec_still_dedupes(client: Client): assert [r.value for r in await read_all(stream.read())] == ["a"] -async def test_an_activity_reads_the_chain_and_writes_to_its_run(client: Client): +async def test_an_activity_handle_names_the_chain_of_its_run(client: Client): streams_client = await connect_with(client, MemoryStreams()) env = ActivityEnvironment(client=streams_client) - async def addresses() -> tuple[str, str]: - inner = activity_handle()._inner - return ( - inner._address("output").run_id, - inner._address("output", producer=True).run_id, + async def chain_run_id() -> str | None: + return activity_handle().ref.chain_run_id + + assert await env.run(chain_run_id) == ( + ActivityEnvironment.default_info().workflow_run_id + ) + + +@workflow.defn +class ContinuesThenWaits: + def __init__(self) -> None: + self.done = False + + @workflow.run + async def run(self, runs_left: int) -> None: + if runs_left: + workflow.continue_as_new(runs_left - 1) + await workflow.wait_condition(lambda: self.done) + + @workflow.signal + def finish(self) -> None: + self.done = True + + +async def wait_for_run_after(client: Client, workflow_id: str, run_id: str) -> None: + async def continued() -> None: + while ( + await client.get_workflow_handle(workflow_id).describe() + ).run_id == run_id: + await asyncio.sleep(0.05) + + await asyncio.wait_for(continued(), 10) + + +async def test_a_read_on_a_named_chain_follows_continue_as_new(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker(streams_client, ContinuesThenWaits) as worker: + handle = await streams_client.start_workflow( + ContinuesThenWaits.run, 1, id=workflow_id, task_queue=worker.task_queue + ) + first_run = handle.first_execution_run_id or "" + await wait_for_run_after(streams_client, workflow_id, first_run) + stream = get_stream_handle( + streams_client, workflow_id, topic=TOKENS, chain_run_id=first_run + ) + producer = stream.producer(producer_id="p", attempt=1) + await producer.append("x") + reader = stream.read() + first = await asyncio.wait_for(anext(reader), 10) + waiting = asyncio.ensure_future(anext(reader)) + # Past Core's first owner check while the read waits. A read pinned to + # the named run would end there, since that run closed at + # Continue-as-New. + await asyncio.sleep(1.5) + await producer.append("y") + second = await asyncio.wait_for(waiting, 10) + await streams_client.get_workflow_handle(workflow_id).signal( + ContinuesThenWaits.finish ) + rest = await read_all(reader) + assert [first.value, second.value] == ["x", "y"] + assert rest == [] + - read, write = await env.run(addresses) - assert (read, write) == ("", ActivityEnvironment.default_info().workflow_run_id) +async def test_a_named_chain_stays_apart_from_a_newer_chain(client: Client): + streams_client = await connect_with(client, MemoryStreams()) + workflow_id = new_workflow_id() + async with new_worker(streams_client, ContinuesThenWaits) as worker: + old = await streams_client.start_workflow( + ContinuesThenWaits.run, 0, id=workflow_id, task_queue=worker.task_queue + ) + old_run = old.first_execution_run_id or "" + await ( + get_stream_handle(streams_client, workflow_id, chain_run_id=old_run) + .producer(producer_id="p", attempt=1) + .append("old") + ) + await old.signal(ContinuesThenWaits.finish) + await old.result() + new = await streams_client.start_workflow( + ContinuesThenWaits.run, 0, id=workflow_id, task_queue=worker.task_queue + ) + await ( + get_stream_handle(streams_client, workflow_id) + .producer(producer_id="p", attempt=1) + .append("new") + ) + # The old chain ended, so its read delivers what it holds and ends. + old_records = await read_all( + get_stream_handle(streams_client, workflow_id, chain_run_id=old_run).read() + ) + await new.signal(ContinuesThenWaits.finish) + await new.result() + new_records = await read_all(get_stream_handle(streams_client, workflow_id).read()) + assert [r.value for r in old_records] == ["old"] + assert [r.value for r in new_records] == ["new"] async def test_an_activity_with_no_workflow_has_no_workflow_stream():