From e7002fab1fd002960d527423232a1c84c42a5267 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 15:01:53 -0700 Subject: [PATCH 1/4] Refused stream readers in Workflows and non-Workflow owners. Only Workflows own streams and only Activities and clients read them in this release, so the rest raises StreamUnsupportedError with a clear message. --- temporalio/contrib/streams/__init__.py | 11 ++- temporalio/contrib/streams/_handles.py | 13 ++-- temporalio/contrib/streams/_ref.py | 14 +++- temporalio/contrib/streams/_workflow.py | 24 +++++- tests/contrib/streams/test_outside_handles.py | 15 ---- .../streams/test_unsupported_owners.py | 74 +++++++++++++++++++ 6 files changed, 126 insertions(+), 25 deletions(-) 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 3368546c97..3aa7f97e23 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -42,6 +42,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 record on the wire is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.contrib.streams.proto.v1`, with the user's value in ``body`` as an ordinary payload. @@ -89,7 +93,11 @@ resolve_topic, topic, ) -from temporalio.contrib.streams._workflow import WorkflowStreamWriter, workflow_writer +from temporalio.contrib.streams._workflow import ( + WorkflowStreamWriter, + workflow_reader, + workflow_writer, +) __all__ = [ "ActivityStreamHandle", @@ -125,5 +133,6 @@ "get_stream_handle", "resolve_topic", "topic", + "workflow_reader", "workflow_writer", ] diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 21d13b9ea6..14567e9924 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -15,6 +15,7 @@ import temporalio.activity from temporalio.client import Client from temporalio.contrib.streams._cursor import BEGINNING +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._plugin import ( provider_for_activity, provider_for_client, @@ -118,15 +119,17 @@ def activity_handle() -> ActivityStreamHandle: producer id of your own. Raises: - ValueError: The Activity was not scheduled by a Workflow, or no stream - provider is registered on the Worker or its client. + StreamUnsupportedError: The Activity was not scheduled by a Workflow; + streams owned by an Activity are not supported in this release. + ValueError: No stream provider is registered on the Worker or its + client. RuntimeError: Not called from inside an Activity. """ info = temporalio.activity.info() if info.workflow_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" ) provider = provider_for_activity() ref = StreamRef.for_workflow(info.workflow_id, run_id=info.workflow_run_id) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index 0cd5d02588..34b86fdd77 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, @@ -24,6 +25,8 @@ StreamOwnerKind = Literal["workflow"] """What owns a stream. Only a Workflow does in this release.""" +_LATER_KINDS = ("activity", "standalone") + @dataclass(frozen=True) class StreamRef: @@ -48,10 +51,17 @@ def __post_init__(self) -> None: """Refuse a ref this release cannot open. Raises: - ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or - ``topic`` is empty. + StreamUnsupportedError: ``kind`` names an owner kind that a later + release adds. + ValueError: ``kind`` is unknown, or ``workflow_id`` or ``topic`` + is empty. """ if self.kind != "workflow": + if self.kind in _LATER_KINDS: + raise StreamUnsupportedError( + f"streams owned by an {self.kind!r} are not supported in this " + "release; only Workflow-owned streams are" + ) raise ValueError(f"unknown StreamRef kind {self.kind!r}") if not self.workflow_id: raise ValueError("a Workflow StreamRef needs a workflow_id") diff --git a/temporalio/contrib/streams/_workflow.py b/temporalio/contrib/streams/_workflow.py index 79471447c8..db0e25c22a 100644 --- a/temporalio/contrib/streams/_workflow.py +++ b/temporalio/contrib/streams/_workflow.py @@ -1,7 +1,8 @@ """Publishing to a Workflow's own stream from Workflow code. The writer converts values on the Workflow thread and hands the records to -the provider's output for the run. Nothing here does I/O. +the provider's output for the run. Nothing here does I/O. Reading a stream +inside a Workflow is not part of this release. """ from __future__ import annotations @@ -10,12 +11,13 @@ from typing import Any, Generic, TypeVar, overload from temporalio import workflow +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._plugin import _WorkflowOutput, provider_for_workflow from temporalio.contrib.streams._record import RecordKind from temporalio.contrib.streams._topic import StreamTopic, resolve_topic from temporalio.contrib.streams._wire import to_wire -__all__ = ["WorkflowStreamWriter", "workflow_writer"] +__all__ = ["WorkflowStreamWriter", "workflow_reader", "workflow_writer"] T = TypeVar("T") @@ -157,3 +159,21 @@ def workflow_writer( """ name, _ = resolve_topic(topic) return WorkflowStreamWriter(_run_streams(), name) + + +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 2d2f0ee115..2d912b35b9 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -8,7 +8,6 @@ from __future__ import annotations import asyncio -import dataclasses import uuid from datetime import timedelta from typing import Any @@ -28,7 +27,6 @@ ) from temporalio.contrib.streams.memory import MemoryStreams from temporalio.exceptions import ApplicationError -from temporalio.testing import ActivityEnvironment from tests.contrib.streams.test_workflow_writer import EVENTS, Publisher from tests.helpers import new_worker @@ -191,16 +189,3 @@ async def test_a_client_producer_and_the_workflow_share_a_topic(client: Client): ("", {"step": "init"}), ("", None), ] - - -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) diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py new file mode 100644 index 0000000000..98f9698a5b --- /dev/null +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -0,0 +1,74 @@ +"""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 uuid + +import pytest + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.streams import ( + StreamRef, + StreamUnsupportedError, + activity_handle, + topic, + workflow_reader, +) +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.testing import ActivityEnvironment +from tests.helpers import new_worker + +EVENTS = topic("events", dict) + + +def client_with(client: Client, provider: MemoryStreams) -> Client: + config = client.config() + config["plugins"] = [provider] + return Client(**config) + + +def test_a_ref_refuses_owner_kinds_this_release_lacks(): + for kind in ("activity", "standalone"): + with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): + StreamRef(kind, "x") # type: ignore[arg-type] + + +@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 = client_with(client, MemoryStreams()) + async with new_worker(streams_client, TriesToRead) as worker: + said = await streams_client.execute_workflow( + TriesToRead.run, + id=f"streams-unsupported-{uuid.uuid4().hex}", + 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 b34af5cc62c06e43d95a55f7d2a2e42f656b11ad Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 13:16:14 -0700 Subject: [PATCH 2/4] Refused an unsupported ref kind where a handle opens, not at decode. A ref of a later release that arrives as Workflow input failed while it decoded, which fails the Workflow Task on every retry. --- temporalio/contrib/streams/_handles.py | 1 + temporalio/contrib/streams/_ref.py | 40 +++++++++++-------- temporalio/contrib/streams/memory.py | 5 +++ tests/contrib/streams/test_topics_refs.py | 17 ++++---- .../streams/test_unsupported_owners.py | 5 ++- 5 files changed, 42 insertions(+), 26 deletions(-) diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index 6df22e36c8..9ba3a8e9a0 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -178,4 +178,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 provider_for_client(client).get_stream_handle(client, ref) diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py index 711d11b610..b9e3014948 100644 --- a/temporalio/contrib/streams/_ref.py +++ b/temporalio/contrib/streams/_ref.py @@ -26,12 +26,9 @@ """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. """ -_LATER_KINDS = ("activity", "standalone") - @dataclass(frozen=True) class StreamRef: @@ -56,26 +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: - StreamUnsupportedError: ``kind`` names an owner kind that a later - release adds. - ValueError: ``kind`` is unknown, 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": - if self.kind in _LATER_KINDS: - raise StreamUnsupportedError( - f"streams owned by an {self.kind!r} are not supported in this " - "release; only Workflow-owned streams are" - ) - 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/memory.py b/temporalio/contrib/streams/memory.py index eedb0e78f9..23e4c0c6bb 100644 --- a/temporalio/contrib/streams/memory.py +++ b/temporalio/contrib/streams/memory.py @@ -514,7 +514,12 @@ def get_stream_handle( ``client`` may be ``None`` here, unlike on a storage provider. Then the handle uses the default data converter and the ``default`` namespace, and a read waits until the caller stops it. + + Raises: + StreamUnsupportedError: ``ref`` names an owner kind this release + lacks. """ + ref._require_supported() return MemoryStreamHandle(self, client, ref) async def close(self) -> None: diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 0e4482f09d..8c2d4b3c7b 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -15,6 +15,7 @@ from temporalio.contrib.streams._topic import ( resolve_topic, ) +from temporalio.contrib.streams.memory import MemoryStreams from temporalio.converter import DataConverter OUT = topic("out", dict) @@ -52,24 +53,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_unsupported(): - # The kind is a plain string, so the converter builds the ref and the ref - # says what is missing, 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(), ) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref.kind == kind with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): - converter.from_payloads([payload], [StreamRef]) + MemoryStreams().get_stream_handle(None, ref) @pytest.mark.parametrize( diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py index 98f9698a5b..e4e9c5dfa2 100644 --- a/tests/contrib/streams/test_unsupported_owners.py +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -33,10 +33,11 @@ def client_with(client: Client, provider: MemoryStreams) -> Client: return Client(**config) -def test_a_ref_refuses_owner_kinds_this_release_lacks(): +def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens(): for kind in ("activity", "standalone"): + ref = StreamRef(kind, "x") # type: ignore[arg-type] with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): - StreamRef(kind, "x") # type: ignore[arg-type] + MemoryStreams().get_stream_handle(None, ref) @workflow.defn From 4a3bf5df372324a6a03c5ecfa8c345d56b6eefbd Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:09:00 -0700 Subject: [PATCH 3/4] Refused unsupported owners on the Core-backed handles. The handles and the writer were rewritten over Core, so the refusals of other owner kinds and of a Workflow read move onto the new code. --- temporalio/contrib/streams/__init__.py | 4 ++++ temporalio/contrib/streams/_handles.py | 12 +++++----- temporalio/contrib/streams/_workflow.py | 22 ++++++++++++++++++- tests/contrib/streams/test_outside_handles.py | 2 -- tests/contrib/streams/test_topics_refs.py | 3 +-- .../streams/test_unsupported_owners.py | 20 ++++++++--------- 6 files changed, 42 insertions(+), 21 deletions(-) diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index e0595259dd..0c8b0e7203 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -44,6 +44,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. diff --git a/temporalio/contrib/streams/_handles.py b/temporalio/contrib/streams/_handles.py index bc16b7ab30..d4719cfbb3 100644 --- a/temporalio/contrib/streams/_handles.py +++ b/temporalio/contrib/streams/_handles.py @@ -443,15 +443,16 @@ def activity_handle() -> ActivityStreamHandle: producer id of your own. Raises: - ValueError: The Activity was not scheduled by a Workflow, or no stream - store is registered on the Worker's client. + StreamUnsupportedError: The Activity was not scheduled by a Workflow; + 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() 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) @@ -497,4 +498,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/_workflow.py b/temporalio/contrib/streams/_workflow.py index 12834bc655..44d9748321 100644 --- a/temporalio/contrib/streams/_workflow.py +++ b/temporalio/contrib/streams/_workflow.py @@ -4,7 +4,8 @@ records to the activation's ``WorkflowOutputStreamCommit``. Nothing here does I/O. The Worker's completion encoder runs each body through the payload codec, and Core stages the records, records a marker of them, and makes them -visible once the Workflow Task is accepted. +visible once the Workflow Task is accepted. Reading a stream inside a +Workflow is not part of this release. """ from __future__ import annotations @@ -16,6 +17,7 @@ from temporalio.bridge.proto.streams.v1 import StreamRecordKind from temporalio.bridge.proto.workflow_commands import OutputRecord from temporalio.contrib.streams._body import content_hash +from temporalio.contrib.streams._errors import StreamUnsupportedError from temporalio.contrib.streams._plugin import worker_has_store from temporalio.contrib.streams._topic import StreamTopic, resolve_topic @@ -156,3 +158,21 @@ def workflow_writer( """ name, _ = resolve_topic(topic) return WorkflowStreamWriter(name) + + +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 bc991a0f36..b497a21632 100644 --- a/tests/contrib/streams/test_outside_handles.py +++ b/tests/contrib/streams/test_outside_handles.py @@ -7,7 +7,6 @@ from __future__ import annotations -import dataclasses import uuid from collections.abc import Sequence from datetime import timedelta @@ -35,7 +34,6 @@ from temporalio.contrib.streams.memory import MemoryStreams from temporalio.converter import DataConverter, PayloadCodec 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.contrib.streams.test_workflow_writer import EVENTS, Publisher from tests.helpers import new_worker diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py index 1a5687cd66..509d8cbbf1 100644 --- a/tests/contrib/streams/test_topics_refs.py +++ b/tests/contrib/streams/test_topics_refs.py @@ -15,7 +15,6 @@ from temporalio.contrib.streams._topic import ( resolve_topic, ) -from temporalio.contrib.streams.memory import MemoryStreams from temporalio.converter import DataConverter OUT = topic("out", dict) @@ -72,7 +71,7 @@ def test_a_ref_from_a_later_release_decodes_and_fails_where_it_opens(kind: str): ref = converter.from_payloads([payload], [StreamRef])[0] assert ref.kind == kind with pytest.raises(StreamUnsupportedError, match="only Workflow-owned"): - MemoryStreams().get_stream_handle(None, ref) + ref._require_supported() @pytest.mark.parametrize( diff --git a/tests/contrib/streams/test_unsupported_owners.py b/tests/contrib/streams/test_unsupported_owners.py index e4e9c5dfa2..498957f430 100644 --- a/tests/contrib/streams/test_unsupported_owners.py +++ b/tests/contrib/streams/test_unsupported_owners.py @@ -7,7 +7,6 @@ from __future__ import annotations import dataclasses -import uuid import pytest @@ -17,27 +16,26 @@ 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) -def client_with(client: Client, provider: MemoryStreams) -> Client: - config = client.config() - config["plugins"] = [provider] - return Client(**config) - - -def test_a_ref_refuses_owner_kinds_this_release_lacks_where_it_opens(): +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"): - MemoryStreams().get_stream_handle(None, ref) + get_stream_handle(streams_client, ref) @workflow.defn @@ -52,11 +50,11 @@ async def run(self) -> str: async def test_a_workflow_cannot_read_a_stream(client: Client): - streams_client = client_with(client, MemoryStreams()) + 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=f"streams-unsupported-{uuid.uuid4().hex}", + id=new_workflow_id(), task_queue=worker.task_queue, ) assert "not supported in this release" in said From 9b472f720420b050e5f1dca7c7ae979b6b372991 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:40:58 -0700 Subject: [PATCH 4/4] Restored the Activity test imports the P3b merge dropped. Both branches touched the import block, and the merge kept P3c's, which lost the imports the Local Activity case uses. --- tests/contrib/streams/test_outside_handles.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/contrib/streams/test_outside_handles.py b/tests/contrib/streams/test_outside_handles.py index 7c92b8b12c..c91e5abcb2 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 dataclasses import uuid from collections.abc import Sequence from datetime import timedelta @@ -41,6 +42,7 @@ 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.contrib.streams.test_workflow_writer import EVENTS, Publisher from tests.helpers import new_worker